netty篇-第一个netty服务器

netty篇-第一个netty服务器

先了解几个概念：Channel 、 回调、Future、事件和ChannelHandler

放在一起之后：

1. Future、回调、ChannelHandler

2. 选择器、事件、EventLoop

echo服务器的交互方式

编写ChannelHandler和业务逻辑

@ChannelHandler.Sharable//表示一个handler可以被多个channel安全的共享
public class EchoServerHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf in = (ByteBuf) msg;
        System.out.println("Server received: " + in.toString(CharsetUtil.UTF_8));
        ctx.write(in);//将接收到的消息发给发送者，而不冲刷出站消息
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)//将未解决的消息冲刷到远程节点
                .addListener(ChannelFutureListener.CLOSE);//完成之后新增一个关闭channel的监听器
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();//如果消息接受出现异常，打印异常信息
        ctx.close();//关闭该channel，防止占用资源
    }
}

编写引导服务器

public class EchoServer {

    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws InterruptedException {
        if (args.length != 1) {
            System.out.println("Usage: " + EchoServer.class.getSimpleName() +
                    " <port>");//设置端口值，如果端口参数格式不正确，打印启动帮助信息
        }
        int port = Integer.parseInt(args[0]);
        new EchoServer(port).start();//调用服务器的start方法，开启服务器
    }

    private void start() throws InterruptedException {
        final EchoServerHandler echoServerHandler = new EchoServerHandler();//准备好服务器端的入站处理器
        EventLoopGroup group = new NioEventLoopGroup();//准备好处理消息需要使用的eventLoop组
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();//开始创建启动引导程序
            bootstrap.group(group)//设置eventLoop组
                    .channel(NioServerSocketChannel.class)//设置nio通信管道
                    .localAddress(new InetSocketAddress(port))//设置服务启动绑定地址
                    .childHandler(new ChannelInitializer<SocketChannel>() {//将自定义的入栈处理器绑定到处理器列表的最后
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(echoServerHandler);
                        }
                    });
            ChannelFuture future = bootstrap.bind()//绑定之后
                    .sync();//同步
            // 这个语句的主要目的是，如果缺失上述代码，则main方法所在的线程，即主线程会在执行完bind().sync()方法后，
            // 会进入finally 代码块，之前的启动的netty server也会随之关闭掉，整个程序都结束了。
            future.channel()
                    .closeFuture()
                    .sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }
}

编写echo客户端的业务处理逻辑

@ChannelHandler.Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {//记录已接受消息的转储
        //todo 可以做一些消息的转储，消息被读取的时候调用
        System.out.println("Client received: " + msg.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {//当被通知channel是活跃的时候，发送一条消息
        //新连接被建立的时候调用
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

编写echo客户端

public class EchoClient {
    private final String host;
    private final int port;

    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();//每个客户端启动都需要先创建一个eventLoop组
        try {
            Bootstrap bootstrap = new Bootstrap();//创建客户端的启动器
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)//设置channel类型
                    .remoteAddress(new InetSocketAddress(host, port))
                    .handler(new ChannelInitializer<SocketChannel>() {//设置channel初始化方法
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new EchoClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect().sync();//连接同步
            future.channel().closeFuture().sync();//为防止程序关闭资源直接被释放，这里加个关闭同步
        } finally {
            group.shutdownGracefully().sync();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        if (args.length != 2) {
            System.out.println("Usage: " + EchoClient.class.getSimpleName() +
                    " <host> <port>");
        }
        String host = args[0];
        int port = Integer.parseInt(args[1]);
        new EchoClient(host, port).start();
    }
}

直接启动