RocketMQ源码：NameServer 消息处理

偏执的太偏执、 2024-03-26 20:04 76阅读 0赞

## 1. 处理业务请求的`ChannelHandler`：serverHandler ##

`NamesrvController`启动后，就可以处理`Broker/Producer/Consumer`的请求消息了，处理该类型消息的`ChannelHandler`为`serverHandler`，也就是`NettyRemotingServer.NettyServerHandler`（`NettyServerHandler`是`NettyRemotingServer`的内部类），代码如下：

class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) 
                throws Exception {
            processMessageReceived(ctx, msg);
        }
    }

继续跟进`NettyRemotingAbstract#processMessageReceived`方法：

public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
        final RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                // 请求消息
                case REQUEST_COMMAND:
                    // todo
                    processRequestCommand(ctx, cmd);
                    break;
                // 响应消息
                case RESPONSE_COMMAND:
                    // todo
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

这里我们直接跟进`REQUEST_COMMAND`命令的处理方法`NettyRemotingAbstract#processRequestCommand`：

public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
        // 根据 code 从 processorTable 获取 Pair
        final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
        // 找不到默认值
        final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
        final int opaque = cmd.getOpaque();
    
        if (pair != null) {
            Runnable run = new Runnable() {
                @Override
                public void run() {
                    try {
                        doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
                        final RemotingResponseCallback callback = new RemotingResponseCallback() {
                            @Override
                            public void callback(RemotingCommand response) {
                                doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd, response);
                                // 不是 单向
                                if (!cmd.isOnewayRPC()) {
                                    if (response != null) {
                                        response.setOpaque(opaque);
                                        response.markResponseType();
                                        try {
                                            ctx.writeAndFlush(response);
                                        } catch (Throwable e) {
                                            log.error("process request over, but response failed", e);
                                            log.error(cmd.toString());
                                            log.error(response.toString());
                                        }
                                    } else {
                                    }
                                }
                            }
                        };
                        // 异步netty请求处理器
                        if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
                            AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor)pair.getObject1();
                            processor.asyncProcessRequest(ctx, cmd, callback);
                        } else {
                            // 不是 异步请求处理器 从 pair 中拿到 Processor 进行处理
                            NettyRequestProcessor processor = pair.getObject1();
                            // todo 处理请求
                            RemotingCommand response = processor.processRequest(ctx, cmd);
                            callback.callback(response);
                        }
                    } 
                    ...
                }
            };
    
            if (pair.getObject1().rejectRequest()) {
                final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                    "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }
    
            try {
                final RequestTask requestTask = new RequestTask(run, ctx.channel(), cmd);
                pair.getObject2().submit(requestTask);
            } catch (RejectedExecutionException e) {
               ...
            }
        } else {
            ...
        }
    }

这个方法主要流程为，先获取`Pair`对象，然后将处理操作封装为`Runnable`对象，接着把`Runnable`对象提交到线程池中。

这个`Pair`对象是啥呢？还记得我们在`NamesrvController#initialize`方法中创建的`remotingExecutor`吗，它最终注册到为`NettyRemotingServer`的`defaultRequestProcessor`属性：

@Override
    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executor) {
        this.defaultRequestProcessor 
                = new Pair<NettyRequestProcessor, ExecutorService>(processor, executor);
    }

这里获取的`Pair`对象正是`defaultRequestProcessor`，`pair.getObject2()`得到的线程池正是`remotingExecutor`，`pair.getObject1()`得到的`processor`是`DefaultRequestProcessor`.

这里我们就明白了，`remotingExecutor`线程池就是用来处理远程请求的。

远程命令的处理逻辑在`Runnable#run`方法中：

public void run() {
        try {
            doBeforeRpcHooks(RemotingHelper.parseChannelRemoteAddr(ctx.channel()), cmd);
            ...
            // 处理请求
            if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
                AsyncNettyRequestProcessor proc