Seata 核心类分析

[luky116]

ResourceManager：资源管理器

ResourceManager（RM，资源管理器）主要负责分布式事务生命周期的所有活动，比如：

• 向TC注册事务（branchRegister）
• 提交事务（branchCommit）
• 回滚事务（branchRollback）
• 向TC注册事务方法信息（registerResource）

每一种模式都会定义一个自己的RM，比如上面的就是TCC模式的RM，它继承了AbstractResourceManager抽象类，这个抽象类里面定义了通用的 事务分支注册、事务状态报告、资源注册等逻辑，每种事务模式要自定义分支提交和回滚的逻辑。这里利用了模板设计模式。
在Seata中，RM是通过SPI的机制来加载的。

RMHandler：RM对外统一的操作入口

每种事务方式都对应一个RMHandler类，通过装饰器模式，里面封装了一个ResourceManager类，RMHandler的所有操作实际上是调用了ResourceManager类的逻辑来完成的。

�这样做的好处是，通过SPI机制加载ResourceManager，自动化了RMHandler的配置流程，用户使用的时候，只需要通过SPI配置好RM即可使用。

public class RMHandlerTCC extends AbstractRMHandler {
    @Override
    protected ResourceManager getResourceManager() {
        // 通过SPI机制获取到ResourceManager
        return DefaultResourceManager.get().getResourceManager(BranchType.TCC);
    }
    @Override
    public BranchType getBranchType() {
        return BranchType.TCC;
    }
}

注意，AbstractRMHandler 类的下面两个方法，是调用ResourceManager类实现的

@Override
public BranchCommitResponse handle(BranchCommitRequest request) {
  BranchCommitResponse response = new BranchCommitResponse();
  exceptionHandleTemplate(new AbstractCallback<BranchCommitRequest, BranchCommitResponse>() {
    @Override
    public void execute(BranchCommitRequest request, BranchCommitResponse response)
      throws TransactionException {
      // 这里实际是调用了ResourceManager来操作的
      // getResourceManager().branchCommit(request.getBranchType(), xid, branchId, resourceId, applicationData);
      doBranchCommit(request, response);
    }
  }, request, response);
  return response;
}

@Override
public BranchRollbackResponse handle(BranchRollbackRequest request) {
  BranchRollbackResponse response = new BranchRollbackResponse();
  exceptionHandleTemplate(new AbstractCallback<BranchRollbackRequest, BranchRollbackResponse>() {
    @Override
    public void execute(BranchRollbackRequest request, BranchRollbackResponse response)
      throws TransactionException {
      // 这里实际是调用了ResourceManager来操作的
      // getResourceManager().branchRollback(request.getBranchType(), xid, branchId, resourceId, applicationData);
      doBranchRollback(request, response);
    }
  }, request, response);
  return response;
    }

�

DefaultResourceManager

DefaultResourceManager 是单例模式，他里面根据BranchType来动态选择RM执行器，然后执行RM的实际逻辑来执行。这里使用了装饰器模式。

protected static Map<BranchType, ResourceManager> resourceManagers
        = new ConcurrentHashMap<>();

@Override
public void registerResource(Resource resource) {
    getResourceManager(resource.getBranchType()).registerResource(resource);
}

registerResource 方法会进入到这里执行，这里的话就直接调用了Netty的逻辑，来向TC发送请求了：

@Override
public void registerResource(Resource resource) { RmNettyRemotingClient.getInstance().registerResource(resource.getResourceGroupId(), resource.getResourceId());
}

registerResource 继续往下走，会走到这里的逻辑：
管理TC侧的链接信息

public void registerResource(String resourceGroupId, String resourceId) {

    // Resource registration cannot be performed until the RM client is initialized
    if (StringUtils.isBlank(transactionServiceGroup)) {
        return;
    }

    // 这里的channel是从配置文件里面读取的
    if (getClientChannelManager().getChannels().isEmpty()) {
        getClientChannelManager().reconnect(transactionServiceGroup);
        return;
    }
    synchronized (getClientChannelManager().getChannels()) {
        for (Map.Entry<String, Channel> entry : getClientChannelManager().getChannels().entrySet()) {
            String serverAddress = entry.getKey();
            Channel rmChannel = entry.getValue();
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info("will register resourceId:{}", resourceId);
            }
            // 具体代码参考下面
            sendRegisterMessage(serverAddress, rmChannel, resourceId);
        }
    }
}

public void sendRegisterMessage(String serverAddress, Channel channel, String resourceId) {
        RegisterRMRequest message = new RegisterRMRequest(applicationId, transactionServiceGroup);
        message.setResourceIds(resourceId);
        try {
            // 发送TC请求
            super.sendAsyncRequest(channel, message);
        } catch (FrameworkException e) {
            if (e.getErrcode() == FrameworkErrorCode.ChannelIsNotWritable && serverAddress != null) {
                getClientChannelManager().releaseChannel(channel, serverAddress);
                if (LOGGER.isInfoEnabled()) {
                    LOGGER.info("remove not writable channel:{}", channel);
                }
            } else {
                LOGGER.error("register resource failed, channel:{},resourceId:{}", channel, resourceId, e);
            }
        }
    }

sendRegisterMessage 里面调用了io.seata.core.rpc.netty.AbstractNettyRemotingClient类里面的 sendAsyncRequest 方法

NettyRemotingClient-TM侧发送/处理消息的核心类

�

AbstractNettyRemoting 类的processMessage是处理消息的入口方法：

protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getBody()));
        }
        Object body = rpcMessage.getBody();
        if (body instanceof MessageTypeAware) {
            MessageTypeAware messageTypeAware = (MessageTypeAware) body;
            final Pair<RemotingProcessor, ExecutorService> pair = 
                // 根据typeCode来选择消息的处理器
                this.processorTable.get((int) messageTypeAware.getTypeCode());
            if (pair != null) {
                if (pair.getSecond() != null) {
                    try {
                        pair.getSecond().execute(() -> {
                            try {
                                pair.getFirst().process(ctx, rpcMessage);
                            } catch (Throwable th) {
                                LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                            } finally {
                                MDC.clear();
                            }
                        });
                    } catch (RejectedExecutionException e) {
                        LOGGER.error(FrameworkErrorCode.ThreadPoolFull.getErrCode(),
                            "thread pool is full, current max pool size is " + messageExecutor.getActiveCount());
                        if (allowDumpStack) {
                            String name = ManagementFactory.getRuntimeMXBean().getName();
                            String pid = name.split("@")[0];
                            int idx = new Random().nextInt(100);
                            try {
                                Runtime.getRuntime().exec("jstack " + pid + " >d:/" + idx + ".log");
                            } catch (IOException exx) {
                                LOGGER.error(exx.getMessage());
                            }
                            allowDumpStack = false;
                        }
                    }
                } else {
                    try {
                        pair.getFirst().process(ctx, rpcMessage);
                    } catch (Throwable th) {
                        LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                    }
                }
            } else {
                LOGGER.error("This message type [{}] has no processor.", messageTypeAware.getTypeCode());
            }
        } else {
            LOGGER.error("This rpcMessage body[{}] is not MessageTypeAware type.", body);
        }
    }

NettyRemotingServer：TC侧处理消息逻辑

RemotingProcessor-消息处理器逻辑

为了方便看，截图大图放这里：

注册中心-RegistryService

public interface RegistryService<T> {

    /**
     * The constant PREFIX_SERVICE_MAPPING.
     */
    String PREFIX_SERVICE_MAPPING = "vgroupMapping.";
    /**
     * The constant PREFIX_SERVICE_ROOT.
     */
    String PREFIX_SERVICE_ROOT = "service";
    /**
     * The constant CONFIG_SPLIT_CHAR.
     */
    String CONFIG_SPLIT_CHAR = ".";

    Set<String> SERVICE_GROUP_NAME = new HashSet<>();

    /**
     * Register.
     *
     * @param address the address
     * @throws Exception the exception
     */
    void register(InetSocketAddress address) throws Exception;

    /**
     * Unregister.
     *
     * @param address the address
     * @throws Exception the exception
     */
    void unregister(InetSocketAddress address) throws Exception;

    /**
     * Subscribe.
     *
     * @param cluster  the cluster
     * @param listener the listener
     * @throws Exception the exception
     */
    void subscribe(String cluster, T listener) throws Exception;

    /**
     * Unsubscribe.
     *
     * @param cluster  the cluster
     * @param listener the listener
     * @throws Exception the exception
     */
    void unsubscribe(String cluster, T listener) throws Exception;

    /**
     * Lookup list.
     *
     * @param key the key
     * @return the list
     * @throws Exception the exception
     */
    List<InetSocketAddress> lookup(String key) throws Exception;

    /**
     * Close.
     * @throws Exception
     */
    void close() throws Exception;

    /**
     * Get current service group name
     *
     * @param key service group
     * @return the service group name
     */
    default String getServiceGroup(String key) {
        key = PREFIX_SERVICE_ROOT + CONFIG_SPLIT_CHAR + PREFIX_SERVICE_MAPPING + key;
        if (!SERVICE_GROUP_NAME.contains(key)) {
            ConfigurationCache.addConfigListener(key);
            SERVICE_GROUP_NAME.add(key);
        }
        return ConfigurationFactory.getInstance().getConfig(key);
    }
}

以下是目前支持的所有注册中心方式：

管理TC侧的连接-ClientChannelManager

管理Session - SessionManager

TC侧消息处理类：

�