RocketMQ通信协议

Posted 2023-04-04 Think_Higher

文章目录

• 通讯协议
• • Header data
  • • code
    • language
    • version
    • opaque
    • flag
    • remark
    • extFields

RocketMQ的通信协议其实很简单，但是无论是官方的用户手册，还是网上的博客，并没有很清晰简单地把其中所有的内容和原理讲明白。 对于需要扩展其他语言SDK的开发来说，意味着必须要深入到Java源码才能弄懂其概念。

注：现最新版RocketMQ已经可以支持Protobuf协议，但本文只探究默认的基于JSON的协议。

无论是发送消息，拉取消息，还是发送心跳等所有的网络通讯层协议（客户端与broker/nameserver间，broker与nameserver间）都使用一套一样的协议。并且无论请求还是响应，协议是一样的，协议头的字段也是固定的。

通讯协议

协议分为以下四部分：

RocketMQ Remoting模块源码功能分析

基础通信模块

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
                            );
                    
                );