seata数据源代理
阅读原文时间:2023年07月08日阅读:1

seata数据源代理流程

1-SeataDataSourceAutoConfiguration

创建SeataAutoDataSourceProxyCreator对象,默认seata模式为AT

2-SeataAutoDataSourceProxyCreator

  • 设置advisor用于来接实现DataSource接口的实现类

    public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes, String dataSourceProxyMode) {
          setProxyTargetClass(!useJdkProxy);
          this.excludes = new HashSet<>(Arrays.asList(excludes));
          // AT
          this.dataSourceProxyMode = dataSourceProxyMode;
          this.advisors = buildAdvisors(dataSourceProxyMode);
      }
    
      private Object[] buildAdvisors(String dataSourceProxyMode) {
          Advice advice = new SeataAutoDataSourceProxyAdvice(dataSourceProxyMode);
          return new Object[]{new DefaultIntroductionAdvisor(advice)};
      }
  • 重写wrapIfNecessary,设置数据源代理

      @Override
      protected Object wrapIfNecessary(Object bean, String beanName, Object cacheKey) {
          // we only care DataSource bean
          if (!(bean instanceof DataSource)) {
              return bean;
          }
      // when this bean is just a simple DataSource, not SeataDataSourceProxy
      LOGGER.error("bean={}", bean.getClass().getSimpleName());
      if (!(bean instanceof SeataDataSourceProxy)) {
          Object enhancer = super.wrapIfNecessary(bean, beanName, cacheKey);
          // this mean this bean is either excluded by user or had been proxy before
          if (bean == enhancer) {
              return bean;
          }
          // else, build proxy,  put &lt;origin, proxy&gt; to holder and return enhancer
          DataSource origin = (DataSource) bean;
          SeataDataSourceProxy proxy = buildProxy(origin, dataSourceProxyMode);
          DataSourceProxyHolder.put(origin, proxy);
          LOGGER.error("bean2origin={}, proxy={}, proxy.targetDataSource={}", origin.getClass().getSimpleName(), proxy.getClass().getSimpleName(), proxy.getTargetDataSource());
          return enhancer;
      }
    
      /*
       * things get dangerous when you try to register SeataDataSourceProxy bean by yourself!
       * if you insist on doing so, you must make sure your method return type is DataSource,
       * because this processor will never return any subclass of SeataDataSourceProxy
       */
      LOGGER.warn("Manually register SeataDataSourceProxy(or its subclass) bean is discouraged! bean name: {}", beanName);
      SeataDataSourceProxy proxy = (SeataDataSourceProxy) bean;
      DataSource origin = proxy.getTargetDataSource();
      LOGGER.error("TargetDataSource={}", origin.getClass().getSimpleName());
      Object originEnhancer = super.wrapIfNecessary(origin, beanName, cacheKey);
      // this mean origin is either excluded by user or had been proxy before
      if (origin == originEnhancer) {
          return origin;
      }
      // else, put &lt;origin, proxy&gt; to holder and return originEnhancer
      DataSourceProxyHolder.put(origin, proxy);
      return originEnhancer;
    }

3-SeataAutoDataSourceProxyAdvice

所有实现DataSource接口的实现类都会被拦截而调用invoke方法。

此处真是被调用的方法是DataSourceProxy#getConnection(),该方法会返回connection代理对象ConnectionProxy。

数据库操作初始化流程:DataSource->Connection->PrepareStatement

  @Override
  public Object invoke(MethodInvocation invocation) throws Throwable {
      // check whether current context is expected
      if (!inExpectedContext()) {
          return invocation.proceed();
      }

      Method method = invocation.getMethod();
      String name = method.getName();
      Class<?>[] parameterTypes = method.getParameterTypes();

      Method declared;
      try {
          declared = DataSource.class.getDeclaredMethod(name, parameterTypes);
      } catch (NoSuchMethodException e) {
          // mean this method is not declared by DataSource
          return invocation.proceed();
      }

      // switch invoke instance to its proxy
      DataSource origin = (DataSource) invocation.getThis();
      SeataDataSourceProxy proxy = DataSourceProxyHolder.get(origin);
      Object[] args = invocation.getArguments();
      LOGGER.error("执行类={},执行方法={},方法参数={}", proxy.getClass().getSimpleName(), declared.getName(), args);
      return declared.invoke(proxy, args);
  }

4-AbstractConnectionProxy

在上一步操作中,会得到connection代理对象ConnectionProxy,数据库SQL在执行前还需要初始化PreparedStatement,此时会调用ConnectionProxy父类AbstractConnectionProxy#prepareStatement(sql)方法,从而得到PreparedStatement代理类PreparedStatementProxy。

  @Override
  public PreparedStatement prepareStatement(String sql) throws SQLException {
      String dbType = getDbType();
      // support oracle 10.2+
      PreparedStatement targetPreparedStatement = null;
      if (BranchType.AT == RootContext.getBranchType()) {
          List<SQLRecognizer> sqlRecognizers = SQLVisitorFactory.get(sql, dbType);
          if (sqlRecognizers != null && sqlRecognizers.size() == 1) {
              SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);
              if (sqlRecognizer != null && sqlRecognizer.getSQLType() == SQLType.INSERT) {
                  TableMeta tableMeta = TableMetaCacheFactory.getTableMetaCache(dbType).getTableMeta(getTargetConnection(), sqlRecognizer.getTableName(), getDataSourceProxy().getResourceId());
                  String[] pkNameArray = new String[tableMeta.getPrimaryKeyOnlyName().size()];
                  tableMeta.getPrimaryKeyOnlyName().toArray(pkNameArray);
                  targetPreparedStatement = getTargetConnection().prepareStatement(sql, pkNameArray);
              }
          }
      }
      if (targetPreparedStatement == null) {
          targetPreparedStatement = getTargetConnection().prepareStatement(sql);
      }
      return new PreparedStatementProxy(this, targetPreparedStatement, sql);
  }

5-PreparedStatementProxy

SQL在执行时会调用PreparedStatementProxy#execute方法

    @Override
    public boolean execute() throws SQLException {
        // 执行两次
        // 第一次:代理类PreparedStatementProxy
        // 第二次:被代理的真实的PreparedStatement,即:targetStatement,此时执行业务SQL
        return ExecuteTemplate.execute(this, (statement, args) -> statement.execute());
    }

    @Override
    public ResultSet executeQuery() throws SQLException {
        return ExecuteTemplate.execute(this, (statement, args) -> statement.executeQuery());
    }

    @Override
    public int executeUpdate() throws SQLException {
        return ExecuteTemplate.execute(this, (statement, args) -> statement.executeUpdate());
    }

6-ConnectionProxy

SQL在执行完成后,最后一步会执行commit方法

    @Override
    public void commit() throws SQLException {
        try {
            lockRetryPolicy.execute(() -> {
                doCommit();
                return null;
            });
        } catch (SQLException e) {
            if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
                rollback();
            }
            throw e;
        } catch (Exception e) {
            throw new SQLException(e);
        }
    }