基础通信模块
Remoting模块是RocketMQ的基础通信模块。
分布式应用中不可避免的是跨进程通信，此类问题可以通过RPC调用解决，RocketMQ的Producer、Consumer、Broker之间的通讯也是通过RPC实现的，高性能RPC调用的主题：传输、协议、线程。
需要约定好特定的通讯协议。 消息传输完成后，通过什么样的线程模型处理线程请求也很重要。
传输，IO通信模型决定了通信性能，RocketMQ的remoting模块通过Netty实现了IO多路复用的Reactor通信模型。
在NameServer初始化完成后启动时，会创建一个NettyRemotingServer对象赋值给remotingServer，而后在Start方法中启动NettyRemotingServer的一个Netty服务端并初始化一个channel。
流程图示
NamesrvStartup.class
代码语言：javascript复制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
代码语言：javascript复制 public boolean initialize() {
this.kvConfigManager.load();
//初始化NettyRemotingServer
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
}NettyRemotingServer.class
代码语言：javascript复制 @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
);
}
});
