Seata解析

文章23 |   阅读 13401 |   点赞0

来源：https://blog.csdn.net/weixin_38308374/category_10261111.html

Seata解析-seata核心类NettyRemotingServer详解

转x33g5p2x  于2021-12-21 转载在 其他  
字(13.0k)|赞(0)|评价(0)|浏览(531)

本文基于seata 1.3.0版本

NettyRemotingServer是TM服务端的核心类之一，它用于启动netty，监听服务器端口，接收TM、RM的请求，它还为处理不同的请求创建不同的处理器。本文将详解NettyRemotingServer类。

一、NettyRemotingServer继承体系

下图是NettyRemotingServer的继承图，里面展示的方法都是protected及以上的：

从上图可以看出，NettyRemotingServer实现了Disposable和RemotingServer接口，继承了AbstractNettyRemotingServer和AbstractNettyRemoting类。
Disposable接口只有一个destroy方法，该方法是在JVM停止的时候，回调钩子调用。
RemotingServer接口定义了发送请求和响应信息的方法以及一个注册处理器的方法，从名字上可以看出这个接口是用于服务端的。
AbstractNettyRemoting抽象类定义了同步和异步发送消息方法（sendSync和sendAsync）以及将收到的消息转交给对应处理器处理的方法（processMessage），该类定义发送消息和接受消息的基础方法，该类是服务端和客户端实现与Netty相关的远程通讯的基类。
AbstractNettyRemotingServer实现了RemotingServer接口，继承了AbstractNettyRemoting抽象类，该类完成下面几个功能：

  • 当发送消息时，该类会调用AbstractNettyRemoting中对应的方法进行发送；
  • 实现了registerProcessor处理器注册方法，可以注册处理器，处理器注册到该类的属性processorTable中；
  • 定义了内部类ServerHandler，它实现自Netty的ChannelDuplexHandler，ServerHandler会被注册到Netty的责任链中，因此当收到消息时，所有的消息都会进入到ServerHandler的channelRead方法中处理；
  • 该类还有一个属性NettyServerBootstrap，NettyServerBootstrap的作用是用于启动和停止服务端Netty，该类的init方法调用NettyServerBootstrap的start方法启动服务端netty。

NettyRemotingServer的init方法有两个作用：一是调用registerProcessor注册处理器，二是调用父类的init方法启动服务端netty。
服务端主要是接收消息并对其进行处理，因此需要重点关注启动流程和接收消息流程，下图展示了启动和接收消息场景中关键方法的调用关系：

二、接收消息处理器

服务端接收的消息最终都是由处理器处理的，那么seata服务端一共注册了哪些处理器？
服务端的处理器都是通过NettyRemotingServer的registerProcessor方法注册的，该方法在服务端启动的时候由NettyRemotingServer的init方法调用：

private void registerProcessor() {
        // 1. registry on request message processor
        ServerOnRequestProcessor onRequestProcessor =
            new ServerOnRequestProcessor(this, getHandler());
        super.registerProcessor(MessageType.TYPE_BRANCH_REGISTER, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_BRANCH_STATUS_REPORT, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_BEGIN, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_COMMIT, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_LOCK_QUERY, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_REPORT, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_ROLLBACK, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_GLOBAL_STATUS, onRequestProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_SEATA_MERGE, onRequestProcessor, messageExecutor);
        // 2. registry on response message processor
        ServerOnResponseProcessor onResponseProcessor =
            new ServerOnResponseProcessor(getHandler(), getFutures());
        super.registerProcessor(MessageType.TYPE_BRANCH_COMMIT_RESULT, onResponseProcessor, messageExecutor);
        super.registerProcessor(MessageType.TYPE_BRANCH_ROLLBACK_RESULT, onResponseProcessor, messageExecutor);
        // 3. registry rm message processor
        RegRmProcessor regRmProcessor = new RegRmProcessor(this);
        super.registerProcessor(MessageType.TYPE_REG_RM, regRmProcessor, messageExecutor);
        // 4. registry tm message processor
        RegTmProcessor regTmProcessor = new RegTmProcessor(this);
        super.registerProcessor(MessageType.TYPE_REG_CLT, regTmProcessor, null);
        // 5. registry heartbeat message processor
        ServerHeartbeatProcessor heartbeatMessageProcessor = new ServerHeartbeatProcessor(this);
        super.registerProcessor(MessageType.TYPE_HEARTBEAT_MSG, heartbeatMessageProcessor, null);
    }

上面的代码注册处理器时，都调用了父类的registerProcessor方法：

//第一个参数是消息类型，
	//第二个参数是处理器对象，
	//第三个参数messageExecutor是线程池，
	//如果messageExecutor不为null，那么在处理消息时会在线程池中增加一个处理器处理消息的异步任务，对消息做异步处理，如果messageExecutor为null，那么消息会同步处理
	public void registerProcessor(int messageType, RemotingProcessor processor, ExecutorService executor) {
        Pair<RemotingProcessor, ExecutorService> pair = new Pair<>(processor, executor);
        //processorTable的类型是HashMap<Integer, Pair<RemotingProcessor, ExecutorService>>，
        //当收到消息的时候，会根据消息类型从processorTable找到对应的处理器和线程池
        this.processorTable.put(messageType, pair);
    }

处理器一共有五个，每个处理器处理的消息类型如下表：

这五个处理器都实现了接口RemotingProcessor，该接口也非常简单，只有一个process方法：

public interface RemotingProcessor {

    void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception;
}

seata的客户端发送消息的时候，会将消息类型添加到消息中，这样接收端在接收到消息后，便可以找到对应的类，解析为对应的消息对象，最后消息对象被转发给AbstractNettyRemoting的processMessage，processMessage里面根据消息对象中的消息类型从processorTable中找到对应的处理器，之后处理器在将消息对象做进一步处理，processMessage方法的代码如下：

//下面将部分代码做了删减
	protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        Object body = rpcMessage.getBody();
        //seata消息都是MessageTypeAware的
        if (body instanceof MessageTypeAware) {
            MessageTypeAware messageTypeAware = (MessageTypeAware) body;
            //根据消息类型从processorTable中找到对应的处理器和线程池
            final Pair<RemotingProcessor, ExecutorService> pair = this.processorTable.get((int) messageTypeAware.getTypeCode());
            if (pair != null) {
            	//根据线程池是否为null，处理消息分为同步和异步
                if (pair.getSecond() != null) {
                    try {
                        pair.getSecond().execute(() -> {
                            try {
                            	//异步调用处理器的process方法
                                pair.getFirst().process(ctx, rpcMessage);
                            } catch (Throwable th) {
                                LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
                            }
                        });
                    } catch (RejectedExecutionException e) {
                        ...//代码删减
                } else {
                    try {
                    	//同步调用处理器的process方法
                        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);
        }
    }

下面具体看一下每个处理器的实现。

1、ServerOnRequestProcessor

本处理接收到消息后，进行一些简单的校验，校验通过的消息会转发给协调器DefaultCoordinator处理做进一步处理。

public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        //检查当前的channel是否注册过，
        //在发起事务请求前，TM和RM需要发送注册信息在服务器端注册，服务器端注册是通过RegRmProcessor和RegTmProcessor处理的
        //seata中的连接都是长连接，TM、RM和服务端建立连接后，会一直使用相同的channel通讯
        if (ChannelManager.isRegistered(ctx.channel())) {
            onRequestMessage(ctx, rpcMessage);
        } else {
            //如果当前channel没有注册过，seata任务是非法连接，直接关闭该连接。
            try {
                if (LOGGER.isInfoEnabled()) {
                    LOGGER.info("closeChannelHandlerContext channel:" + ctx.channel());
                }
                ctx.disconnect();
                ctx.close();
            } catch (Exception exx) {
                LOGGER.error(exx.getMessage());
            }
            if (LOGGER.isInfoEnabled()) {
                LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
            }
        }
    }

    private void onRequestMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
        Object message = rpcMessage.getBody();
        //获取RpcContext上下文对象
        //RpcContext上下文对象是在TM和RM注册的时候创建的
        RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
        //下面的if/else分支用于打印日志，从这可以看出，seata的日志打印功能是比较完备的
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("server received:{},clientIp:{},vgroup:{}", message,
                NetUtil.toIpAddress(ctx.channel().remoteAddress()), rpcContext.getTransactionServiceGroup());
        } else {
            try {
                BatchLogHandler.INSTANCE.getLogQueue()
                    .put(message + ",clientIp:" + NetUtil.toIpAddress(ctx.channel().remoteAddress()) + ",vgroup:"
                        + rpcContext.getTransactionServiceGroup());
            } catch (InterruptedException e) {
                LOGGER.error("put message to logQueue error: {}", e.getMessage(), e);
            }
        }
        //消息类型不是AbstractMessage直接丢弃
        if (!(message instanceof AbstractMessage)) {
            return;
        }
        //MergedWarpMessage表示当前的消息是一个复合消息，里面包含了多个消息，需要对每个消息进行遍历
        if (message instanceof MergedWarpMessage) {
            AbstractResultMessage[] results = new AbstractResultMessage[((MergedWarpMessage) message).msgs.size()];
            for (int i = 0; i < results.length; i++) {
                final AbstractMessage subMessage = ((MergedWarpMessage) message).msgs.get(i);
                //调用协调器DefaultCoordinator处理
                results[i] = transactionMessageHandler.onRequest(subMessage, rpcContext);
            }
            MergeResultMessage resultMessage = new MergeResultMessage();
            resultMessage.setMsgs(results);
            //异步发送响应消息
            remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), resultMessage);
        } else {
            // the single send request message
            final AbstractMessage msg = (AbstractMessage) message;
            //调用协调器DefaultCoordinator处理
            AbstractResultMessage result = transactionMessageHandler.onRequest(msg, rpcContext);
            //异步发送响应消息
            remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), result);
        }
    }

2、ServerOnResponseProcessor

本处理器的对响应消息的处理分为两部分：一是需要对响应消息做特殊处理的，则通过触发回调逻辑完成对响应消息的处理；二是使用协调器DefaultCoordinator完成处理。

public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        //发送的消息如果需要对响应消息做特殊处理的都会记录到futures中，
        //这样当响应消息到达时，可以通过MessageFuture对象触发回调逻辑对响应消息处理
        MessageFuture messageFuture = futures.remove(rpcMessage.getId());
        if (messageFuture != null) {
            messageFuture.setResultMessage(rpcMessage.getBody());
        } else {
            if (ChannelManager.isRegistered(ctx.channel())) {
                onResponseMessage(ctx, rpcMessage);
            } else {
                //如果当前channel没有注册过，seata任务是非法连接，直接关闭该连接。
                try {
                    if (LOGGER.isInfoEnabled()) {
                        LOGGER.info("closeChannelHandlerContext channel:" + ctx.channel());
                    }
                    ctx.disconnect();
                    ctx.close();
                } catch (Exception exx) {
                    LOGGER.error(exx.getMessage());
                }
                if (LOGGER.isInfoEnabled()) {
                    LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
                }
            }
        }
    }

    private void onResponseMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
        //打印日志使用
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("server received:{},clientIp:{},vgroup:{}", rpcMessage.getBody(),
                NetUtil.toIpAddress(ctx.channel().remoteAddress()),
                ChannelManager.getContextFromIdentified(ctx.channel()).getTransactionServiceGroup());
        } else {
            try {
                BatchLogHandler.INSTANCE.getLogQueue()
                    .put(rpcMessage.getBody() + ",clientIp:" + NetUtil.toIpAddress(ctx.channel().remoteAddress()) + ",vgroup:"
                        + ChannelManager.getContextFromIdentified(ctx.channel()).getTransactionServiceGroup());
            } catch (InterruptedException e) {
                LOGGER.error("put message to logQueue error: {}", e.getMessage(), e);
            }
        }
        if (rpcMessage.getBody() instanceof AbstractResultMessage) {
            //获取RpcContext上下文对象
            RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
            //调用协调器DefaultCoordinator处理
            transactionMessageHandler.onResponse((AbstractResultMessage) rpcMessage.getBody(), rpcContext);
        }
    }

3、RegRmProcessor

本处理器主要是调用ChannelManager.registerRMChannel方法将RM信息注册到服务端。registerRMChannel方法的具体实现逻辑后续文章在做介绍。

public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        onRegRmMessage(ctx, rpcMessage);
    }

    private void onRegRmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
        //得到RM注册请求消息
        RegisterRMRequest message = (RegisterRMRequest) rpcMessage.getBody();
        //获得请求方的ip和端口，用于打印日志使用
        String ipAndPort = NetUtil.toStringAddress(ctx.channel().remoteAddress());
        boolean isSuccess = false;
        String errorInfo = StringUtils.EMPTY;
        try {
            //在本版本里面权限检测处理器checkAuthHandler的功能还不完善，权限检测直接返回true
            if (null == checkAuthHandler || checkAuthHandler.regResourceManagerCheckAuth(message)) {
                //RM信息注册到服务器端
                ChannelManager.registerRMChannel(message, ctx.channel());
                //存储客户端的版本号，从目前的版本代码看，只是做了存储，可能在后续版本会使用
                Version.putChannelVersion(ctx.channel(), message.getVersion());
                isSuccess = true;
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("checkAuth for client:{},vgroup:{},applicationId:{} is OK", ipAndPort, message.getTransactionServiceGroup(), message.getApplicationId());
                }
            }
        } catch (Exception exx) {
            isSuccess = false;
            errorInfo = exx.getMessage();
            LOGGER.error("RM register fail, error message:{}", errorInfo);
        }
        //构建响应消息
        RegisterRMResponse response = new RegisterRMResponse(isSuccess);
        if (StringUtils.isNotEmpty(errorInfo)) {
            response.setMsg(errorInfo);
        }
        //异步发送响应消息
        remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("RM register success,message:{},channel:{},client version:{}", message, ctx.channel(),
                message.getVersion());
        }
    }

4、RegTmProcessor

本处理器主要是调用ChannelManager.registerTMChannel方法将TM信息注册到服务端。registerTMChannel方法的具体实现逻辑后续文章在做介绍。

public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        onRegTmMessage(ctx, rpcMessage);
    }

    private void onRegTmMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
        //得到TM注册请求消息
        RegisterTMRequest message = (RegisterTMRequest) rpcMessage.getBody();
        //获得请求方的ip和端口，用于打印日志使用
        String ipAndPort = NetUtil.toStringAddress(ctx.channel().remoteAddress());
        //存储客户端的版本号，从目前的版本代码看，只是做了存储，可能在后续版本会使用
        Version.putChannelVersion(ctx.channel(), message.getVersion());
        boolean isSuccess = false;
        String errorInfo = StringUtils.EMPTY;
        try {
            //在本版本里面权限检测处理器checkAuthHandler的功能还不完善，权限检测直接返回true
            if (null == checkAuthHandler || checkAuthHandler.regTransactionManagerCheckAuth(message)) {
                //TM信息注册到服务器端
                ChannelManager.registerTMChannel(message, ctx.channel());
                Version.putChannelVersion(ctx.channel(), message.getVersion());
                isSuccess = true;
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("checkAuth for client:{},vgroup:{},applicationId:{}",
                        ipAndPort, message.getTransactionServiceGroup(), message.getApplicationId());
                }
            }
        } catch (Exception exx) {
            isSuccess = false;
            errorInfo = exx.getMessage();
            LOGGER.error("TM register fail, error message:{}", errorInfo);
        }
        //构建响应消息
        RegisterTMResponse response = new RegisterTMResponse(isSuccess);
        if (StringUtils.isNotEmpty(errorInfo)) {
            response.setMsg(errorInfo);
        }
        //异步发送响应消息
        remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
        if (LOGGER.isInfoEnabled()) {
            LOGGER.info("TM register success,message:{},channel:{},client version:{}", message, ctx.channel(),
                message.getVersion());
        }
    }

5、ServerHeartbeatProcessor

心跳处理器的处理逻辑还是非常简单的，收到心跳消息后，直接同步返回响应。

public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
        try {
            remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), HeartbeatMessage.PONG);
        } catch (Throwable throwable) {
            LOGGER.error("send response error: {}", throwable.getMessage(), throwable);
        }
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("received PING from {}", ctx.channel().remoteAddress());
        }
    }

后面的文章会对协调器和ChannelManager做进一步的分析。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/weixin_38308374/article/details/108401885

内容来源于网络，如有侵权，请联系作者删除！

Javarpc分布式事务seatasping boot

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com