0
点赞
收藏
分享

微信扫一扫

RocketMQ之通信模块(三)

冶炼厂小练 2022-01-09 阅读 6

 通信机制

以上是官方关于RocketMQ通讯机制的解释。需要详细了解其原理可以参考https://github.com/apache/rocketmq/blob/master/docs/cn/design.md

Remoting通信类

通信类结构:

 NettyRemotingServer

  • NettyRemotingServer为服务端实现类,在NamesrvController中被构造。NettyRemotingServer构造时主要工作是初始化下列属性,构造时判断useEpoll来决定EventLoopGroup的实现。
    private final ServerBootstrap serverBootstrap;
    private final EventLoopGroup eventLoopGroupSelector;
    private final EventLoopGroup eventLoopGroupBoss;
    private final NettyServerConfig nettyServerConfig;

    private final ExecutorService publicExecutor;
    private final ChannelEventListener channelEventListener;
  • NettyRemotingServer启动
        @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();
    
            ServerBootstrap childHandler =
                this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
                    .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .option(ChannelOption.SO_REUSEADDR, true)
                    .option(ChannelOption.SO_KEEPALIVE, false)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                    .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                    .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
    					//添加handler,握手、编解码、idle检测、连接管理、消息处理
                            ch.pipeline()
                                .addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME, handshakeHandler)
                                .addLast(defaultEventExecutorGroup,
                                    encoder,
                                    new NettyDecoder(),
                                    new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                    connectionManageHandler,
                                    serverHandler
                                );
                        }
                    });
    
            if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
                childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            }
    
            try {
                ChannelFuture sync = this.serverBootstrap.bind().sync();
                InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
                this.port = addr.getPort();
            } catch (InterruptedException e1) {
                throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
            }
    
            if (this.channelEventListener != null) {
                this.nettyEventExecutor.start();
            }
    
            this.timer.scheduleAtFixedRate(new TimerTask() {
    
                @Override
                public void run() {
                    try {
                        NettyRemotingServer.this.scanResponseTable();
                    } catch (Throwable e) {
                        log.error("scanResponseTable exception", e);
                    }
                }
            }, 1000 * 3, 1000);
        }

    主要关注initChannel方法中添加的handler,NettyConnectManageHandler
     

        //消息处理核心类   
        @ChannelHandler.Sharable
        class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
                processMessageReceived(ctx, msg);
            }
        }
    
        //连接管理处理类
        @ChannelHandler.Sharable
        class NettyConnectManageHandler extends ChannelDuplexHandler {
    
            @Override
            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                log.info("NETTY SERVER PIPELINE: channelActive, the channel[{}]", remoteAddress);
                super.channelActive(ctx);
    
                if (NettyRemotingServer.this.channelEventListener != null) {
                    NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remoteAddress, ctx.channel()));
                }
            }
    
            @Override
            public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                log.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
                super.channelInactive(ctx);
    
                if (NettyRemotingServer.this.channelEventListener != null) {
                    NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
                }
            }
    
            @Override
            public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                if (evt instanceof IdleStateEvent) {
                    IdleStateEvent event = (IdleStateEvent) evt;
                    if (event.state().equals(IdleState.ALL_IDLE)) {
                        final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                        log.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
                        RemotingUtil.closeChannel(ctx.channel());
                        if (NettyRemotingServer.this.channelEventListener != null) {
                            NettyRemotingServer.this
                                .putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
                        }
                    }
                }
    
                ctx.fireUserEventTriggered(evt);
            }
    
            @Override
            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                log.warn("NETTY SERVER PIPELINE: exceptionCaught {}", remoteAddress);
                log.warn("NETTY SERVER PIPELINE: exceptionCaught exception.", cause);
    
                if (NettyRemotingServer.this.channelEventListener != null) {
                    NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
                }
    
                RemotingUtil.closeChannel(ctx.channel());
            }
        }

NettyRemotingClient

举报

相关推荐

0 条评论