Netty 3升级Netty4实践

参考

修改点

1、按标准的启动流程模板将Netty 3替换为Netty 4

2、Handler替换，需要考虑有Netty4的变化，将Handler的功能分析清楚并使用Netty 4的方式实现

3、Netty 3到Netty4的主要修改

ChannelBuffer -> ByteBuf
ChannelBuffers -> PooledByteBufferAllocator (需要注意使用完成后释放buffer)或UnpooledByteBufferAllocator
解码器 FremeDecoder -> ByteToMessageDecoder
编码器 OneToOneEncoder -> MessageToByteEncoder

版本

netty 4的项目结构变化

二进制JAR已拆分为多个子模块，因此用户可以从类路径中排除不必要的功能。当前结构如下：

Artifact ID 描述

netty-parent Maven父POM
netty-common 工具类和日志框架
netty-buffer 替代 java.nio.ByteBuffer的ByteBuf API
netty-transport Channel API和核心传输core transports
netty-transport-rxtx Rxtx传输
netty-transport-sctp SCTP传输
netty-transport-udt UDT传输
netty-handler 有用的ChannelHandler实现
netty-codec 有助于编写编码器和解码器的编解码器框架
netty-codec-http 与HTTP，Web套接字，SPDY和RTSP相关的编解码器
netty-codec-socks 与SOCKS协议相关的编解码器
netty-all 结合了以上所有模块的多合一JAR
netty-tarball Tarball发行版本
netty-example 例子
netty-testsuite-* 集成测试的集合
netty-microbench 微基准

现在，所有Artifacts（除了netty-all.jar）都是OSGi捆绑包，可以在您喜欢的OSGi容器中使用。

目标版本及依赖的artifactId

升级为4.1.47.Final，没有依赖软件和安全漏洞，提供了更丰富的编解码器（包括SMTP）

注意需要引入netty-all，不要单独引用netty的分包

<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.47.Final</version>
</dependency>

Netty中Channel、ChannelPipeline、ChannelHandler、ChannelHandlerContext之间的关系

每一个Channel被创建，就会生成对应的一个ChannelPipeline和它绑定。
ChannelPipeline中包含了一个处理该Channel消息的ChannelHandler链。
当每一个ChannelHandler被注册到该ChannelPipeline中就会生成一个对应的 ChannelHandlerContext，和该ChannelHandler进行绑定。
一个ChannelHandler可以从属于（注册到）多个ChannelPipeline。所以，一个ChannelHandler可以绑定多个ChannelHandlerContext。不过，这样的ChannelHandler必须使用@Sharable注解标注，保证它的线程安全性，否则试图将它注册到多个ChannelHandlerPipeline中时将会抛出异常。

Netty 4修改项

新的 bootstrap API

bootstrap API已经被重写，尽管它的目的还是一样；它执行需要配置和运行服务器或客户端程序的典型步骤，通常能在样板代码里找到。

新的bootstrap同样采取了流式接口(fluent interface)。

核心修改

启动服务器方式修改

    public static void main(String[] args) throws Exception {
        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 100)
                    .localAddress(8080)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(handler1, handler2, ...);
                        }
                    });
            // Start the server.
            ChannelFuture f = b.bind().sync();
            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            // Wait until all threads are terminated.
            bossGroup.terminationFuture().sync();
            workerGroup.terminationFuture().sync();
        }
    }

ChannelPipelineFactory → ChannelInitializer

就像你在在上面的例子注意到的一样，ChannelPipelineFactory 不再存在了。而是由 ChannelInitializer来替换，它给予了Channel 和 ChannelPipeline 配置的更多控制。

ChannelPipeline 不再让用户创建。ChannelPipeline 由 Channel自动创建。

核心修改

设置childHandler时通过如下方法添加pipeline及handler

.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {

ch.pipeline().addLast(handler1, handler2, ...);

}
});

Inbound事件

继承SimpleChannelInboundHandler<I>，覆写channelRead0方法，Netty会自动释放泛型资源资源。

核心修改

服务端需要继承SimpleChannelInboundHandler，同时需要指定泛型类型为

ImapChannelInboundHandler->Imap->SimpleChannelInboundHandler<ImapMessage>

BasicChannelInboundHandler->POP3/SMTP->SimpleChannelInboundHandler< ByteBuf>

注意释放msg

管道中的每个 inbound (a.k.a. upstream) handler 必须release接收到的消息. Netty不会自动release 它们.

核心修改

继承ChannelInboundHandlerAdapter，当我们需要释放ByteBuf相关内存的时候，也可以使用 ReferenceCountUtil#release()。如果继承SimpleChannelInboundHandler，则其会自动释放消息资源。

ChannelHandler 不需要 event object

4.0完全移除了event object，取而代之的是强类型的方法调用。 3.x 包含处理所有事件的handler method如 handleUpstream() 和 handleDownstream(), 但Netty 4.0中每个 event 类型都有它自己的handler method:

// Before:
void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e);
void handleDownstream(ChannelHandlerContext ctx, ChannelEvent e);
 
// After:
void channelRegistered(ChannelHandlerContext ctx);
void channelUnregistered(ChannelHandlerContext ctx);
void channelActive(ChannelHandlerContext ctx);
void channelInactive(ChannelHandlerContext ctx);
void channelRead(ChannelHandlerContext ctx, Object message);
 
void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise);
void connect(
        ChannelHandlerContext ctx, SocketAddress remoteAddress,
        SocketAddress localAddress, ChannelPromise promise);
void disconnect(ChannelHandlerContext ctx, ChannelPromise promise);
void close(ChannelHandlerContext ctx, ChannelPromise promise);
void deregister(ChannelHandlerContext ctx, ChannelPromise promise);
void write(ChannelHandlerContext ctx, Object message, ChannelPromise promise);
void flush(ChannelHandlerContext ctx);
void read(ChannelHandlerContext ctx);

ChannelHandlerContext 也进行了更改也体现了上述变化：

// Before:
ctx.sendUpstream(evt);
 
// After:
ctx.fireChannelRead(receivedMessage);

核心修改

连接读写消息的获取：msg.getMessage直接取msg

读事件传播：ctx.sendUpstream(evt); ->ctx.fireChannelRead(receivedMessage);

简化channel 状态模型

在3.x中，当一个新的Channel被创建并连接成功，至少三个ChannelStateEvent会被触发：channelOpen、channelBound以及channelConnected.当一个Channel关闭时，也至少有三个事件会被触发：channelDisconnected、channelUnbound以及channelClosed.

但是，触发这么多事件的意义并不那么大。更有用的是当一个Channel进入可读或可写的状态时通知用户。

channelOpen, channelBound, 和 channelConnected 被合并到 channelActive. channelDisconnected, channelUnbound, 和 channelClosed 被合并到 channelInactive.

同样Channel.isBound() 和 isConnected() 也被合并为isActive().

需要注意的是，channelRegistered and channelUnregistered 这两个事件与channelOpen and channelClosed并不等。它们是在支持Channel的动态注册、注销以及再注册时被引入的新的状态。

核心修改

channelBound和channelConnected方法的处理修改为channelActive

channelClosed方法的处理修改为channelInactive

channel.setReadable(false); ->channel.config().setAutoRead(false); 设置是否允许通道读参考

线程模型变化-没有 ExecutionHandler 了

它被放入核心代码中。在你往ChannelPipeline增加ChannelHandler 时你可以指定一个EventExecutor, 这样Pipeline总是使用这个EventExecutor来调用这个新增加的 ChannelHandler的handler方法。

核心修改

去掉显式初始化的ExecutionHandler

write() 不会自动 flush

4.0 引入了新的操作 flush() 它可以显示地将Channel输出缓存输出. write()操作并不会自动 flush. 你可以把它想象成java.io.BufferedOutputStream, 除了它工作于消息级这一点.

由于这个改变, 你必须万分小心，写入数据后不要忘了调用 ctx.flush() . 当然你也可以使用一个更直接的方法 writeAndFlush().

核心修改

写入数据后不要忘了调用 ctx.flush()，或直接调用writeAndFlush()

编解码框架

编码解码器框架里有实质性的内部改变, 因为4.0需要一个handler来创建和管理它的buffer然而，从用户角度来看这些变化并不大。

核心编解码类移入到 io.netty.handler.codec 包下

解码器的作用

以IMAP为例，在Netty 3.10.6.Final的程序中，IMAPServer#createPipelineFactory创建的pipeline最后先添加了解码器，再添加了核心处理器

在解码器ImapRequestFrameDecoder中重写了decode方法，最终返回ImapMessage类型，之后消息在pipeline传递到核心处理器ImapChannelUpstreamHandler#messageReceived方法，在接收的消息事件MessageEvent中，通过getMessage方法获取到解码后的类型

ImapMessage message = (ImapMessage) e.getMessage();

如果去掉解码器ImapRequestFrameDecoder，则传递的是默认的消息类型BigEndianHeapChannelBuffer（父类是ChannelBuffer）

核心修改

FrameDecoder 被重新命名为 ByteToMessageDecoder.

OneToOneEncoder和OneToOneDecoder被MessageToMessageEncoder 和 MessageToMessageDecoder 取代.

decode(), decodeLast(), encode() 的方法签名有些许改变以便支持泛型，也移除了一些冗余的参数。

需要重新编写解码器ImapRequestFrameDecoder，使用Netty 4的

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

进行解码

in：需要解码的二进制数据。
List<Object> out：解码后的有效报文列表，我们需要将解码后的报文添加到这个List中。之所以使用一个List表示，是因为考虑到粘包问题，因此入参的in中可能包含多个有效报文。当然，也有可能发生了拆包，in中包含的数据还不足以构成一个有效报文，此时不往List中添加元素即可

参考

之后可以在核心处理器中获取解码后的Message

AttributeMap

为了响应用户需求，您可以将任何对象附加到Channel和ChannelHandlerContext。增加了一个名为AttributeMap的新接口，该接口被Channel和ChannelHandlerContext实现。同时ChannelLocal和Channel.attachment被删除了。当关联Channel被垃圾收集时，这些属性也将被垃圾收集。因此，可以没有显式去掉属性的方法。（关闭channel）

参考

每一个ChannelHandlerContext都有属于自己的上下文，也就说每一个ChannelHandlerContext上如果有AttributeMap都是绑定上下文的，也就说如果A的ChannelHandlerContext中的AttributeMap，B的ChannelHandlerContext是无法读取到的（Attribute<NettyChannel> attr = ctx.attr(NETTY_CHANNEL_KEY); ）

~~但是Channel上的AttributeMap就是大家共享的，每一个ChannelHandler都能获取到（Attribute<NettyChannel> attr = ctx.channel().attr(NETTY_CHANNEL_KEY); ）~~

Netty团队在4.1版本之后，在每一个Channel内部仅仅保留一个map,确保每个key之间的唯一性，因此每一个Channel不需要多个map，因此直接使用Channel的attr添加。

Attribute<NettyChannel> attr = ctx.channel().attr(NETTY_CHANNEL_KEY);

核心修改

private AttributeKey<ImapSession> getAttributeKey(ChannelHandlerContext ctx) {
    return AttributeKey.valueOf(ctx.channel().id().asLongText() + "," + ctx.channel().remoteAddress().toString());
}
// 设置属性
ctx.channel().attr(getAttributeKey(ctx)).set(imapsession);

NettyConstants的attributes属性整改（ChannelLocal）

在NettyConstants中定义ChannelLocal<Object> attributes = new ChannelLocal<>();

属性，通过分析可知，该属性的作用是添加自定义属性到attributes中，在pipleline的handler之间共享，可以使用Netty 4中的AttributeMap绑定属性到channel上。

另，注意对于Netty服务器来说，新来一个连接即建立了一个Channel，每个Channel都新建了一个pipeline与之对应且唯一，在pipeline中是一组handler，且不同的Channel的pipeline和handler都是新创建的，互不干扰。每个handler都有一个上下文ChannelHandlerContext与之唯一对应（handler不是Sharable）。

Channel.attachment整改

attachment也是在channel间共享的数据，可以使用AttributeMap进行替代。

由于attchment原本是DefaultChannelHandlerContext的属性，Netty 4已经去掉。可以通过将需要设置的对象的地址（object.toString()）作为key，对象本身作为内容存入AttributeMap。

由于一个不是@Sharable的ChannelHandler唯一确定一个ChannelHandlerContext，因此可以使用该上下文的地址作为key标识attachment

ByteBuf

Netty 4.x -- ByteBuf

ByteBuf转为String

buf.toString(CharsetUtil.UTF_8)

核心修改

ChannelBuffers. wrappedBuffer修改为Unpooled.wrappedBuffer

channel的关闭

使用如下方式确保缓冲区内容写出后关闭channel，如果直接调用channel.close方法会立即关闭，可能会有数据在缓冲区未写出。

private void bufferFlushOut(Channel channel) {
    if (channel.isActive()) {
        CompositeByteBuf compBuf = Unpooled.compositeBuffer();
        // 初始化一个空的ByteBuf
        ByteBuf heapBuf = Unpooled.buffer(0);
        channel.writeAndFlush(heapBuf).addListener(ChannelFutureListener.CLOSE);
    }
}

迁移服务端的基本点

使用新的bootstrap API重写 FactorialServer.run() 方法.
不再使用 ChannelFactory 。初始化一个 NioEventLoopGroup (一个用来接受连接，其它的用来处理接受后的连接.
重命名 FactorialServerPipelineFactory 为 FactorialServerInitializer. 让它扩展 ChannelInitializer.
不创建一个 ChannelPipeline, 而是通过 Channel.pipeline()得到它.
让 FactorialServerHandler 扩展 ChannelInboundHandlerAdapter.
用channelInactive()替换 channelDisconnected() .
不再使用handleUpstream().
将 messageReceived() 命名为 channelRead()，并相应的调整方法签名.
用 ctx.writeAndFlush()替换 ctx.write() .
让 BigIntegerDecoder 扩展 ByteToMessageDecoder.
让 NumberEncoder 扩展 MessageToByteEncoder.
encode() 不再返回一个buffer. 由ByteToMessageDecoder负责将编码的数据填入到buffer中.

空闲检测handler

ImapIdleStateHandler原来实现IdleStateAwareChannelHandler，Netty 4已经没有该类，需重写。ImapIdleStateHandler类主要作用是检测连接的客户端是否空闲，并执行相应的动作。

Netty 3和Netty 5的心跳机制

玩转Netty -- 从Netty3升级到Netty4

IdleStateAwareChannelHandlerChannelInboundHandlerAdapter

修改继承的实现类cp.addLast("heartbeatHandler", new HeartbeatHandler());


public class ImapIdleStateHandler extends SimpleChannelInboundHandler<ByteBuf> implements NettyConstants {
userEventTriggered
channelRead0
}

类似的ImapHeartbeatHandler也这样处理

IdleStateHandler

IdleStateAwareChannelHandler已经去除，但 IdleStateHandler类还存在，

cp.addLast("idleTimeoutHandler", new IdleStateHandler(getTimer(), getClientIdleTimeout().toMillis(), NO_WRITER_IDLE_TIMEOUT, NO_ALL_IDLE_TIMEOUT, TimeUnit.MILLISECONDS));
修改为
cp.addLast("idleTimeoutHandler", new IdleStateHandler( NO_WRITER_IDLE_TIMEOUT, NO_WRITER_IDLE_TIMEOUT, NO_ALL_IDLE_TIMEOUT, TimeUnit.MILLISECONDS));

编解码器

ImapRequestFrameDecoder

注意点

非@Sharable的handler必须每次new

如果是非@Sharable的handler，每次添加到pipeline的时候必须new出来，否则会报错

is not a @Sharable handler。

主要原因在于，每个连接（Channel）接入服务器的时候都会初始化一个pipeline，如果pipeline中的handler是类成员且在实例化的时候初始化，则只会有一个handler。

io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1

SimpleChannelInboundHandler 它会自动进行一次释放(即引用计数减1).

参考

继承了SimpleChannelInboundHandler的Handler都会自动释放消息资源

继承了SimpleChannelInboundHandler，类会自动释放资源

如果继承的是ChannelInboundHandlerAdapter，则需要自己释放（只能在不需要继续传递的handler释放），如果不释放且继续传递，则会在TailContext中释放（io.netty.channel.DefaultChannelPipeline#onUnhandledInboundMessage(java.lang.Object)）

Netty 4服务端启动

在Netty官方服务器标准样例中，服务器启动最后调用了 f.channel().closeFuture().sync();

该方法会阻塞主线程，导致只能启动一个服务器，因此在代码中不能添加这行代码。

服务器的关闭的两行代码也要放到unbind方法中。

ChannelOption参数详解

1、ChannelOption.SO_BACKLOG

ChannelOption.SO_BACKLOG对应的是tcp/ip协议listen函数中的backlog参数，函数listen(int socketfd,int backlog)用来初始化服务端可连接队列，服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小

2、ChannelOption.SO_REUSEADDR

ChanneOption.SO_REUSEADDR对应于套接字选项中的SO_REUSEADDR，这个参数表示允许重复使用本地地址和端口，比如，某个服务器进程占用了TCP的80端口进行监听，此时再次监听该端口就会返回错误，使用该参数就可以解决问题，该参数允许共用该端口，这个在服务器程序中比较常使用，比如某个进程非正常退出，该程序占用的端口可能要被占用一段时间才能允许其他进程使用，而且程序死掉以后，内核一需要一定的时间才能够释放此端口，不设置SO_REUSEADDR就无法正常使用该端口。

3、ChannelOption.SO_KEEPALIVE

Channeloption.SO_KEEPALIVE参数对应于套接字选项中的SO_KEEPALIVE，该参数用于设置TCP连接，当设置该选项以后，连接会测试链接的状态，这个选项用于可能长时间没有数据交流的连接。当设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文

4、ChannelOption.SO_SNDBUF和ChannelOption.SO_RCVBUF

ChannelOption.SO_SNDBUF参数对应于套接字选项中的SO_SNDBUF，ChannelOption.SO_RCVBUF参数对应于套接字选项中的SO_RCVBUF这两个参数用于操作接收缓冲区和发送缓冲区的大小，接收缓冲区用于保存网络协议站内收到的数据，直到应用程序读取成功，发送缓冲区用于保存发送数据，直到发送成功。

5、ChannelOption.SO_LINGER

ChannelOption.SO_LINGER参数对应于套接字选项中的SO_LINGER,Linux内核默认的处理方式是当用户调用close（）方法的时候，函数返回，在可能的情况下，尽量发送数据，不一定保证会发生剩余的数据，造成了数据的不确定性，使用SO_LINGER可以阻塞close()的调用时间，直到数据完全发送

6、ChannelOption.TCP_NODELAY

ChannelOption.TCP_NODELAY参数对应于套接字选项中的TCP_NODELAY,该参数的使用与Nagle算法有关,Nagle算法是将小的数据包组装为更大的帧然后进行发送，而不是输入一次发送一次,因此在数据包不足的时候会等待其他数据的到了，组装成大的数据包进行发送，虽然该方式有效提高网络的有效负载，但是却造成了延时，而该参数的作用就是禁止使用Nagle算法，使用于小数据即时传输，于TCP_NODELAY相对应的是TCP_CORK，该选项是需要等到发送的数据量最大的时候，一次性发送数据，适用于文件传输。

7、IP_TOS

IP参数，设置IP头部的Type-of-Service字段，用于描述IP包的优先级和QoS选项。

8、ALLOW_HALF_CLOSURE

Netty参数，一个连接的远端关闭时本地端是否关闭，默认值为False。值为False时，连接自动关闭；为True时，触发ChannelInboundHandler的userEventTriggered()方法，事件为ChannelInputShutdownEvent。

Netty的future.channel().closeFuture().sync();到底有什么用？

主线程执行到这里就 wait 子线程结束，子线程才是真正监听和接受请求的，closeFuture()是开启了一个channel的监听器，负责监听channel是否关闭的状态，如果监听到channel关闭了，子线程才会释放，syncUninterruptibly()让主线程同步等待子线程结果

参考1

如果我们不想加f.channel().closeFuture().sync()又想保证程序正常运行怎么办，去掉finally 里面关闭nettyserver的语句即可。

参考2

对javax.servlet-api的依赖

注意项目中依赖javax.servlet-api的scope是provided的，打包的时候不会将该包打到程序中，因此需要确保运行环境中有该包。