Java IO学习笔记八：多路复用到Netty

作者：Grey

原文地址：Java IO学习笔记八：多路复用到Netty

多路复用多线程方式还是有点麻烦，Netty帮我们做了封装，大大简化了编码的复杂度，接下来熟悉一下netty的基本使用。

Netty+最朴素的阻塞的方式来实现一版客户端和服务端通信的代码，然后再重构成Netty官方推荐的写法。

第一步，引入netty依赖包。

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

准备发送端

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

import static java.nio.charset.StandardCharsets.UTF_8;

/**
 * @author <a href="mailto:410486047@qq.com">Grey</a>
 * @since
 */
public class NettyClientSync {
    public static void main(String[] args) throws Exception {
        NioEventLoopGroup thread = new NioEventLoopGroup(1);
        NioSocketChannel client = new NioSocketChannel();
        thread.register(client);
        ChannelPipeline p = client.pipeline();
        p.addLast(new MyInHandler());
        ChannelFuture connect = client.connect(new InetSocketAddress("192.168.205.138", 9090));
        ChannelFuture sync = connect.sync();
        ByteBuf buf = Unpooled.copiedBuffer("hello server".getBytes());
        ChannelFuture send = client.writeAndFlush(buf);
        send.sync();
        sync.channel().closeFuture().sync();
        System.out.println("client over....");
    }

    static class MyInHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) {
            System.out.println("client  register...");
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            System.out.println("client active...");
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            CharSequence str = buf.getCharSequence(0, buf.readableBytes(), UTF_8);
            System.out.println(str);
            ctx.writeAndFlush(buf);
        }
    }
}

这个客户端主要就是给服务端（192.168.205.138：9090）发送数据， 启动一个服务端：

[root@io ~]# nc -l 192.168.205.138 9090

然后启动客户端，服务端可以接收到客户端发来的数据：

[root@io ~]# nc -l 192.168.205.138 9090
hello server

这就是netty实现的一个客户端，再来看服务端的写法：

import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

/**
 * @author <a href="mailto:410486047@qq.com">Grey</a>
 * @since
 */
public class NettyServerSync {
    public static void main(String[] args) throws Exception {
        NioEventLoopGroup thread = new NioEventLoopGroup(1);
        NioServerSocketChannel server = new NioServerSocketChannel();
        thread.register(server);
        ChannelPipeline p = server.pipeline();
        p.addLast(new MyAcceptHandler(thread, new NettyClientSync.MyInHandler()));
        ChannelFuture bind = server.bind(new InetSocketAddress("192.168.205.1",9090));
        bind.sync().channel().closeFuture().sync();
        System.out.println("server close....");
    }

    static class MyAcceptHandler extends ChannelInboundHandlerAdapter {


        private final EventLoopGroup selector;
        private final ChannelHandler handler;

        public MyAcceptHandler(EventLoopGroup thread, ChannelHandler myInHandler) {
            this.selector = thread;
            this.handler = myInHandler;
        }

        @Override
        public void channelRegistered(ChannelHandlerContext ctx) {
            System.out.println("server registered...");
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            SocketChannel client = (SocketChannel) msg;
            ChannelPipeline p = client.pipeline();
            p.addLast(handler);
            selector.register(client);
        }
    }
}

启动这个服务端，然后通过一个客户端来连接这个服务端，并且向这个服务端发送一些数据

[root@io ~]# nc 192.168.205.1 9090
hello 
hello

服务端可以感知到客户端连接并接收到客户端发来的数据

client  register...
client active...
hello

但是，这样的服务端如果再接收一个客户端连接，客户端继续发送一些数据进来，服务端就会报一个错误：

An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
io.netty.channel.ChannelPipelineException: git.snippets.io.netty.NettyClientSync$MyInHandler is not a @Sharable handler, so can't be added or removed multiple times.

原因在这个博客里面说的比较清楚：Netty ChannelHandler使用报错

我们可以发现每当有新的数据可读时都会往这个channel的pipeline里加入handler，这里加的是childHander。值得注意的是，我们初始化的时候这个childHandler都是同一个实例，也就说会导致不同的channel用了同一个handler，这个从netty的设计角度来说是要避免的。因为netty的一大好处就是每一个channel都有自己绑定的eventloop和channelHandler，这样可以保证代码串行执行，不必考虑并发同步的问题。所以才会有checkMultiplicity这个方法来检查这个问题。那该怎么办呢？netty的这段代码：child.pipeline().addLast(childHandler)就是用了同一个handler啊，怎么才能为每一个channel创建不同的handler呢？
很简单，只要写个类继承ChannelInitializer就行了，ChannelInitializer这个类比较特殊，你可以把它想象成是很多channelhandler的集合体，而且这个类就是@Shareable的，继承了这个类之后你可以为每一个channel单独创建handler，甚至是多个handler。

解决方案也很简单，只需要在服务端传入的handler上加上@Sharable注解即可

@ChannelHandler.Sharable
static class MyInHandler extends ChannelInboundHandlerAdapter{
 ...
}

但是对于每次服务端的Handler，如果都要加@Sharable，就会非常不好扩展，Netty里面提供了一个没有任何业务功能的并且标注为@Sharable的类:ChannelInitializer, 每个业务handler只需要重写其initChannel()方法即可,我们可以改造一下NettyClientSync和NettyServerSync的代码，并用Netty推荐的写法来修改。

客户端改成：

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

/**
 * @author <a href="mailto:410486047@qq.com">Grey</a>
 * @since
 */
public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        ChannelFuture fu = bs
                .group(group).channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        pipeline.addLast(new NettyClientSync.MyInHandler());
                    }
                }).connect(new InetSocketAddress("192.168.205.138", 9090));
        Channel client = fu.channel();
        ByteBuf buf = Unpooled.copiedBuffer("Hello Server".getBytes());
        ChannelFuture future = client.writeAndFlush(buf);
        future.sync();
    }
}

启动一个服务端，然后启动上述客户端代码，服务端可以收到信息

[root@io ~]# nc -l 192.168.205.138 9090
Hello Server

接下来改造服务端代码：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

/**
 * @author <a href="mailto:410486047@qq.com">Grey</a>
 * @since
 */
public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup(1);
        ServerBootstrap bs = new ServerBootstrap();
        ChannelFuture bind = bs
                .group(group, group)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioServerSocketChannel) throws Exception {
                        ChannelPipeline pipeline = nioServerSocketChannel.pipeline();
                        pipeline.addLast(new NettyClientSync.MyInHandler());
                    }
                }).bind(new InetSocketAddress("192.168.205.1", 9090));
        bind.sync().channel().closeFuture().sync();
    }
}

启动服务端代码，然后通过客户端连接服务端并发送一些数据：

[root@io ~]# nc 192.168.205.1 9090
sdfasdfas
sdfasdfas

可以正常接收。

源码：Github