专栏首页码匠的流水账聊聊sharding-jdbc的SQLExecutionHook
原创

聊聊sharding-jdbc的SQLExecutionHook

本文主要研究一下sharding-jdbc的SQLExecutionHook

SQLExecutionHook

incubator-shardingsphere-4.0.0-RC1/sharding-core/sharding-core-execute/src/main/java/org/apache/shardingsphere/core/execute/hook/SQLExecutionHook.java

public interface SQLExecutionHook {
    
    /**
     * Handle when SQL execution started.
     * 
     * @param routeUnit route unit to be executed
     * @param dataSourceMetaData data source meta data
     * @param isTrunkThread is execution in trunk thread
     * @param shardingExecuteDataMap sharding execute data map
     */
    void start(RouteUnit routeUnit, DataSourceMetaData dataSourceMetaData, boolean isTrunkThread, Map<String, Object> shardingExecuteDataMap);
    
    /**
     * Handle when SQL execution finished success.
     */
    void finishSuccess();
    
    /**
     * Handle when SQL execution finished failure.
     *
     * @param cause failure cause
     */
    void finishFailure(Exception cause);
}
  • SQLExecutionHook接口定义了start、finishSuccess、finishFailure方法

SPISQLExecutionHook

incubator-shardingsphere-4.0.0-RC1/sharding-core/sharding-core-execute/src/main/java/org/apache/shardingsphere/core/execute/hook/SPISQLExecutionHook.java

public final class SPISQLExecutionHook implements SQLExecutionHook {
    
    private final Collection<SQLExecutionHook> sqlExecutionHooks = NewInstanceServiceLoader.newServiceInstances(SQLExecutionHook.class);
    
    static {
        NewInstanceServiceLoader.register(SQLExecutionHook.class);
    }
    
    @Override
    public void start(final RouteUnit routeUnit, final DataSourceMetaData dataSourceMetaData, final boolean isTrunkThread, final Map<String, Object> shardingExecuteDataMap) {
        for (SQLExecutionHook each : sqlExecutionHooks) {
            each.start(routeUnit, dataSourceMetaData, isTrunkThread, shardingExecuteDataMap);
        }
    }
    
    @Override
    public void finishSuccess() {
        for (SQLExecutionHook each : sqlExecutionHooks) {
            each.finishSuccess();
        }
    }
    
    @Override
    public void finishFailure(final Exception cause) {
        for (SQLExecutionHook each : sqlExecutionHooks) {
            each.finishFailure(cause);
        }
    }
}
  • SPISQLExecutionHook实现了SQLExecutionHook接口;它使用NewInstanceServiceLoader注册了SQLExecutionHook;sqlExecutionHooks集合由NewInstanceServiceLoader.newServiceInstances创建;start方法遍历sqlExecutionHooks,执行其start方法;finishSuccess方法则遍历sqlExecutionHooks,执行其finishSuccess方法;finishFailure方法则遍历sqlExecutionHooks,执行其finishFailure方法

OpenTracingSQLExecutionHook

incubator-shardingsphere-4.0.0-RC1/sharding-opentracing/src/main/java/org/apache/shardingsphere/opentracing/hook/OpenTracingSQLExecutionHook.java

public final class OpenTracingSQLExecutionHook implements SQLExecutionHook {
    
    private static final String OPERATION_NAME = "/" + ShardingTags.COMPONENT_NAME + "/executeSQL/";
    
    private ActiveSpan activeSpan;
    
    private Span span;
    
    @Override
    public void start(final RouteUnit routeUnit, final DataSourceMetaData dataSourceMetaData, final boolean isTrunkThread, final Map<String, Object> shardingExecuteDataMap) {
        if (!isTrunkThread) {
            activeSpan = ((ActiveSpan.Continuation) shardingExecuteDataMap.get(OpenTracingRootInvokeHook.ACTIVE_SPAN_CONTINUATION)).activate();
        }
        span = ShardingTracer.get().buildSpan(OPERATION_NAME)
                .withTag(Tags.COMPONENT.getKey(), ShardingTags.COMPONENT_NAME)
                .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT)
                .withTag(Tags.PEER_HOSTNAME.getKey(), dataSourceMetaData.getHostName())
                .withTag(Tags.PEER_PORT.getKey(), dataSourceMetaData.getPort())
                .withTag(Tags.DB_TYPE.getKey(), "sql")
                .withTag(Tags.DB_INSTANCE.getKey(), routeUnit.getDataSourceName())
                .withTag(Tags.DB_STATEMENT.getKey(), routeUnit.getSqlUnit().getSql())
                .withTag(ShardingTags.DB_BIND_VARIABLES.getKey(), toString(routeUnit.getSqlUnit().getParameters())).startManual();
        
    }
    
    private String toString(final List<Object> parameterSets) {
        return parameterSets.isEmpty() ? "" : String.format("[%s]", Joiner.on(", ").join(parameterSets));
    }
    
    @Override
    public void finishSuccess() {
        span.finish();
        if (null != activeSpan) {
            activeSpan.deactivate();
        }
    }
    
    @Override
    public void finishFailure(final Exception cause) {
        ShardingErrorSpan.setError(span, cause);
        span.finish();
        if (null != activeSpan) {
            activeSpan.deactivate();
        }
    }
}
  • OpenTracingSQLExecutionHook实现了SQLExecutionHook接口,其start方法创建并启动span、activeSpan;finishSuccess及finishFailure方法都会执行span.finish()及activeSpan.deactivate(),只是finishFailure则会标记span的exception信息

SQLExecuteCallback

incubator-shardingsphere-4.0.0-RC1/sharding-core/sharding-core-execute/src/main/java/org/apache/shardingsphere/core/execute/sql/execute/SQLExecuteCallback.java

@RequiredArgsConstructor
public abstract class SQLExecuteCallback<T> implements ShardingGroupExecuteCallback<StatementExecuteUnit, T> {
    
    private final DatabaseType databaseType;
    
    private final boolean isExceptionThrown;
    
    @Override
    public final Collection<T> execute(final Collection<StatementExecuteUnit> statementExecuteUnits, final boolean isTrunkThread,
                                       final Map<String, Object> shardingExecuteDataMap) throws SQLException {
        Collection<T> result = new LinkedList<>();
        for (StatementExecuteUnit each : statementExecuteUnits) {
            result.add(execute0(each, isTrunkThread, shardingExecuteDataMap));
        }
        return result;
    }
    
    private T execute0(final StatementExecuteUnit statementExecuteUnit, final boolean isTrunkThread, final Map<String, Object> shardingExecuteDataMap) throws SQLException {
        ExecutorExceptionHandler.setExceptionThrown(isExceptionThrown);
        DataSourceMetaData dataSourceMetaData = DataSourceMetaDataFactory.newInstance(databaseType, statementExecuteUnit.getStatement().getConnection().getMetaData().getURL());
        SQLExecutionHook sqlExecutionHook = new SPISQLExecutionHook();
        try {
            sqlExecutionHook.start(statementExecuteUnit.getRouteUnit(), dataSourceMetaData, isTrunkThread, shardingExecuteDataMap);
            T result = executeSQL(statementExecuteUnit.getRouteUnit(), statementExecuteUnit.getStatement(), statementExecuteUnit.getConnectionMode());
            sqlExecutionHook.finishSuccess();
            return result;
        } catch (final SQLException ex) {
            sqlExecutionHook.finishFailure(ex);
            ExecutorExceptionHandler.handleException(ex);
            return null;
        }
    }
    
    protected abstract T executeSQL(RouteUnit routeUnit, Statement statement, ConnectionMode connectionMode) throws SQLException;
}
  • SQLExecuteCallback的execute0方法在执行前创建SPISQLExecutionHook,然后调用sqlExecutionHook.start方法,执行成功之后执行sqlExecutionHook.finishSuccess方法,捕获到SQLException则执行sqlExecutionHook.finishFailure方法

小结

SQLExecutionHook接口定义了start、finishSuccess、finishFailure方法;SPISQLExecutionHook实现了SQLExecutionHook接口;它使用NewInstanceServiceLoader注册了SQLExecutionHook;sqlExecutionHooks集合由NewInstanceServiceLoader.newServiceInstances创建;start方法遍历sqlExecutionHooks,执行其start方法;finishSuccess方法则遍历sqlExecutionHooks,执行其finishSuccess方法;finishFailure方法则遍历sqlExecutionHooks,执行其finishFailure方法

doc

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊Elasticsearch的FixedExecutorBuilder

    本文主要研究一下Elasticsearch的FixedExecutorBuilder

    codecraft
  • 聊聊dubbo-go的availableCluster

    dubbo-go-v1.4.2/cluster/cluster_impl/available_cluster.go

    codecraft
  • 聊聊dubbo-go的availableCluster

    dubbo-go-v1.4.2/cluster/cluster_impl/available_cluster.go

    codecraft
  • C语言编程入门之--第六章C语言控制语句

    导读:本章带读者理解什么是控制语句,然后逐个讲解C语言常用的控制语句,含有控制语句的代码量多起来后就要注意写代码的风格了,本章末节都是练习题,大量的练习...

    啊源股
  • 月球相当于北京的几环?

    WolframChina
  • leetcode: 76. Minimum Window Substring

    JNingWei
  • 我,一个自诩牛逼上天的 Node.js 和小程序开发者,今天就教「快应用」好好做人

    知晓君
  • centos7 安装oracle11

    转载请务必注明原创地址为:http://dongkelun.com/2018/05/05/oracleConf/

    董可伦
  • LinkedBlockingQueue

    大学里的混子
  • Eclipse配置Tomcat

    ![第一步](http://ono60m7tl.bkt.clouddn.com/eclipse1.bmp

    爱撒谎的男孩

扫码关注云+社区

领取腾讯云代金券