RocketMQ Remoting模块源码功能分析

关忆北.

发布于 2022-05-05 15:09:17

1890

发布于 2022-05-05 15:09:17

文章被收录于专栏：关忆北.关忆北.关忆北.

基础通信模块

Remoting模块是RocketMQ的基础通信模块。

分布式应用中不可避免的是跨进程通信，此类问题可以通过RPC调用解决，RocketMQ的Producer、Consumer、Broker之间的通讯也是通过RPC实现的，高性能RPC调用的主题：传输、协议、线程。

需要约定好特定的通讯协议。消息传输完成后，通过什么样的线程模型处理线程请求也很重要。

传输，IO通信模型决定了通信性能，RocketMQ的remoting模块通过Netty实现了IO多路复用的Reactor通信模型。

在NameServer初始化完成后启动时，会创建一个NettyRemotingServer对象赋值给remotingServer，而后在Start方法中启动NettyRemotingServer的一个Netty服务端并初始化一个channel。

流程图示

NamesrvStartup.class

NamesrvController controller = createNamesrvController(args);
start(controller);

public static NamesrvController start(final NamesrvController controller) throws Exception {

        if (null == controller) {
            throw new IllegalArgumentException("NamesrvController is null");
        }

  			//初始化
        boolean initResult = controller.initialize();
        if (!initResult) {
            controller.shutdown();
            System.exit(-3);
        }

NamesrvController.class

    public boolean initialize() {

        this.kvConfigManager.load();
				//初始化NettyRemotingServer
        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    }

NettyRemotingServer.class

   @Override
    public void start() {
        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
            nettyServerConfig.getServerWorkerThreads(),
            new ThreadFactory() {

                private AtomicInteger threadIndex = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                }
            });

        prepareSharableHandlers();
				//初始化Netty channel 
        ServerBootstrap childHandler =
            this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, nettyServerConfig.getServerSocketBacklog())
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                            .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
                            .addLast(defaultEventExecutorGroup,
                                encoder,
                                new NettyDecoder(),
                                new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                connectionManageHandler,
                                serverHandler
                            );
                    }
                });