Apache RocketMQ作为阿里开源的一款高性能、高吞吐量的分布式消息中间件，它的通信能力已得到业界的公认。它的通信层是借助于异常网络框架Netty实现的。在开发网络游戏的时候，Netty也常用于游戏服务器或网关的通信层框架，所以，可以通过学习RocketMQ是如何使用Netty框架，从中借鉴一此应用技巧。
在RocketMQ中的rocketmq-remoting项目就是针对网络层封装的项目，其实使用Netty的时候，最主要的就是以下几个部分：
- 线程池的划分
- 是否使用Epoll
- Netty服务启动的相关配置
- Netty内存池的使用。
- 使用共享的Handler
- 服务优雅的关闭
线程池的划分
RocketMQ的服务端Netty启动类是NettyRemotingServer，在这里声明了四个线程池
//这个其实就是netty中的work线程池，默认用来处理Handler方法的调用
private final EventLoopGroup eventLoopGroupSelector;
// Netty的Boss线程
private final EventLoopGroup eventLoopGroupBoss;
// 公共线程池，这里用来处理RocketMQ的业务调用，这个有Netty没有什么关系
private final ExecutorService publicExecutor;
// 用来处理Handler的线程池
private DefaultEventExecutorGroup defaultEventExecutorGroup;
RocketMQ的Netty启动代码
下面是RocketMQ的Netty服务启动代码，如果我们自己想要创建一个Netty服务，直接抄下面的代码，修改一下相关的配置和Handler就可以了：
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 {
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);
}
尽量使用Epoll
epoll是Linux下多路复用IO接口select/poll的增强版本，它能显著减少程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。先不管epoll的原理是什么，这都是底层的实现，只需要知道epoll的效率比其它的方式高就可以，在能使用epoll的情况下，就应该选择使用Netty socket的Epoll模式。从RocketMQ的Netty启动代码来看，在选择使用的ServerSocketChannel时，它使用了一个方法来判断是否使用Epoll。
private boolean useEpoll() {
return RemotingUtil.isLinuxPlatform()
&& nettyServerConfig.isUseEpollNativeSelector()
&& Epoll.isAvailable();
}
nettyServerConfig.isUseEpollNativeSelector可以让使用者在可以使用epoll的环境下，强制不使用epoll。
使用Netty的ByteBuf内存池
使用Netty的内存池可以减少对象的创建和内存分配，进而减少gc的量。在RocketMQ的服务中，默认是netty开启使用netty的内存池的。
if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
}
使用共享的Handler
在netty中，使用Handler处理网络的消息，不管是接收网络消息，还是返回网络消息，都会经过Handler的处理方法。在一个网络连接创建channel时，channel初始化时就会添加相应的Handler。如果每个Handler内都没有共用的对象，那么这些Handler最好标记为共享的，这样可以减少Handler对象的创建。比如RocketMQ的编码Handler
@ChannelHandler.Sharable
public class NettyEncoder extends MessageToByteEncoder<RemotingCommand> {
private static final InternalLogger log = InternalLoggerFactory.getLogger(RemotingHelper.ROCKETMQ_REMOTING);
@Override
public void encode(ChannelHandlerContext ctx, RemotingCommand remotingCommand, ByteBuf out)
throws Exception {
try {
ByteBuffer header = remotingCommand.encodeHeader();
out.writeBytes(header);
byte[] body = remotingCommand.getBody();
if (body != null) {
out.writeBytes(body);
}
} catch (Exception e) {
log.error("encode exception, " + RemotingHelper.parseChannelRemoteAddr(ctx.channel()), e);
if (remotingCommand != null) {
log.error(remotingCommand.toString());
}
RemotingUtil.closeChannel(ctx.channel());
}
}
}
在服务启动时，会只在NettyRemotingServer创建一个NettyEncoder对象实例，所有的channel实例共同使用这个编码实例。
服务优雅的关闭
所谓服务优雅的关闭，是指在服务需要关闭的时候，在关闭之前，需要把任务处理完，而且在收到关闭时，不再接收新的任务。在所有的Netty业务中，有业务相关的线程池就是NettyRemotingServer中创建的四个线程池，所以在关闭服务的时候，只需要关闭这几个线程池即可。并等待线程池中的任务处理完。
在NettyRemotingServer中有一个shutdown()方法。
@Override
public void shutdown() {
try {
if (this.timer != null) {
this.timer.cancel();
}
this.eventLoopGroupBoss.shutdownGracefully();
this.eventLoopGroupSelector.shutdownGracefully();
if (this.nettyEventExecutor != null) {
this.nettyEventExecutor.shutdown();
}
if (this.defaultEventExecutorGroup != null) {
this.defaultEventExecutorGroup.shutdownGracefully();
}
} catch (Exception e) {
log.error("NettyRemotingServer shutdown exception, ", e);
}
if (this.publicExecutor != null) {
try {
this.publicExecutor.shutdown();
} catch (Exception e) {
log.error("NettyRemotingServer shutdown exception, ", e);
}
}
}
这里面调用的shutdownGracefully()就是
@Override
public Future<?> shutdownGracefully() {
return shutdownGracefully(2, 15, TimeUnit.SECONDS);
}
它会等待执行完线程池中当前已所有任务，然后再关闭线程池。
那么，何时调用的这个shutdown()方法呢。在RocketMQ服务启动的时候，会添加一个回调钩子，比如Namesrv服务在启动的时候会执行下面的代码：
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
@Override
public Void call() throws Exception {
controller.shutdown();
return null;
}
}));
这样在服务器关闭的时候，就会触发controller.shudown()。然后执行关闭线程池的操作。
注意，关闭服务器一般使用kill pid的命令，RocketMQ的发布包里面的bin下面，有一个mqshutdown的脚本，就是使用的kill pid 命令。
pid=`ps ax | grep -i 'org.apache.rocketmq.namesrv.NamesrvStartup' |grep java | grep -v grep | awk '{print $1}'`
if [ -z "$pid" ] ; then
echo "No mqnamesrv running."
exit -1;
fi
echo "The mqnamesrv(${pid}) is running..."
kill ${pid}
首先是使用脚本获取进程名的pid,然后使用 kill pid将进程杀死。
但是不能使用kill -9 pid，因为这个命令不会给进程执行回调的机会，就把进程杀死了。
