Netty

NettyClient
NettyClientHandler
NettyServer
NettyServerHandler

package com.example.netty.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class NettyServer {
    public static void main(String[] args) throws InterruptedException {
        // 创建 BossGroup 和 WorkerGroup
        // BossGroup 只处理连接请求，WorkerGroup 处理真正的和客户端的业务处理
        // 两个都是无限循环
        // bossGroup 和 workerGroup 含有的子线程（NioEventLoop）个数默认为实际 CPU 核数 * 2
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try{// 创建服务器端启动对象，配置启动参数
            ServerBootstrap bootstrap = new ServerBootstrap();

            // 使用链式编程进行设置
            bootstrap.group(bossGroup, workerGroup)  // 设置两个线程组
                    .channel(NioServerSocketChannel.class)  // 设置服务器通道实现为 NioServerSocketChannel
                    .option(ChannelOption.SO_BACKLOG, 128)  // 设置线程队列等待连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)  // 设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>() {  // 创建一个通道测试对象，对应的是 workerHandler；而 handler 对应的是 bossGroup
                        // 给 pipeline 设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyServerHandler()); // 向管道内添加处理器
                        }
                    });  // 给 WorkerGroup 的 EventLoop 对应的管道设置处理器
            System.out.println("服务器启动成功，等待客户端连接...");
            // 绑定端口，同步等待成功，生成一个 ChannelFuture 对象（启动服务器，并绑定端口）
            ChannelFuture channelFuture = bootstrap.bind(6668).sync();
            // 等待服务器【监听端口关闭】，通过 ChannelFuture 的 sync() 方法同步等待
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 优雅退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

package com.example.netty.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;


/**
 * 自定义一个 Handler 需要继续 netty 规定好的某个 HandlerAdapter
 * 这时我们自定义一个 Handler，才能成为一个 Handler
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取数据
     * @param ctx 上下文对象，含有管道 pipeline，通道 channel，地址等信息
     * @param msg 客户端发送的数据，默认是 Object
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("server ctx=" + ctx);
        // 将 msg 转为 ByteBuf，注意这是 netty 的 ByteBuf，不是 nio 的 ByteBuffer
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("server received: " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("client address: " + ctx.channel().remoteAddress());

        // 如果有一个非常耗时的业务，会考虑对它异步执行，将它提交到该 channel 对应的 NioEventLoop 的 taskQueue 中
//        ctx.channel().eventLoop().execute(new Runnable() {
//            @Override
//            public void run() {
//                try {
//                    Thread.sleep(10 * 1000);
//                    ctx.writeAndFlush(Unpooled.copiedBuffer("较为复杂的业务发生", CharsetUtil.UTF_8));
//                } catch (InterruptedException e) {
//                    System.out.println("发生异常 " + e.getMessage());
//                }
//            }
//        });

        // 用户自定义定时任务，提交到 scheduleTaskQueue 中
        ctx.channel().eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                ctx.writeAndFlush(Unpooled.copiedBuffer("自定义定时任务", CharsetUtil.UTF_8));
            }
        }, 5, TimeUnit.SECONDS);
    }

    /**
     * 数据读取完毕
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将数据写入到缓冲并刷新
        // 一般需要堆发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端", CharsetUtil.UTF_8));
    }

    /**
     * 异常处理，一半需要关闭通道
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

package com.example.netty.netty.simple;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        // 客户端需要一个时间循环组
        EventLoopGroup group = new NioEventLoopGroup();

        try{// 创建一个客户端启动对象，注意客户端使用的是 BootStrap 而不是 ServerBootStrap
            Bootstrap bootstrap = new Bootstrap();

            // 设置相关参数
            bootstrap.group(group)  // 设置线程组
                    .channel(NioSocketChannel.class)  // 设置客户端通道的实现类（利用反射处理）
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler());  // 加入自己的处理器
                        }
                    });
            System.out.println("客户端启动");
            // 启动客户端去连接服务器端
            // 关于 ChannelFuture 要分析，涉及到异步的概念
            ChannelFuture channelFuture = bootstrap.connect("localhost", 6668).sync();
            // 给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            // 关闭线程组
            group.shutdownGracefully();
        }
    }
}

package com.example.netty.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 当通道就绪就会触发该方法
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(io.netty.channel.ChannelHandlerContext ctx) throws Exception {
        System.out.println("client channelActive: " + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 服务器", CharsetUtil.UTF_8));
    }

    /**
     * 当通道有读取事件时会触发该方法
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息: " + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器地址: "+ctx.channel().remoteAddress());
    }

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

ctx.channel().eventLoop().execute(new Runnable() {
    @Override
    public void run() {
        try {
            Thread.sleep(10 * 1000);
            ctx.writeAndFlush(Unpooled.copiedBuffer("较为复杂的业务发生", CharsetUtil.UTF_8));
        } catch (InterruptedException e) {
            System.out.println("发生异常 " + e.getMessage());
        }
    }
});

ctx.channel().eventLoop().schedule(new Runnable() {
    @Override
    public void run() {
        ctx.writeAndFlush(Unpooled.copiedBuffer("自定义定时任务", CharsetUtil.UTF_8));
    }
}, 5, TimeUnit.SECONDS);

serverBootstrap.bind(port).addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
            System.out.println("端口监听成功");
        } else {
            System.out.println("端口监听失败");
        }
    }
});

// 用于服务器端，用来设置两个 EventLoopGroup
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup);
// 用于客户端，用来设置一个 EventLoopGroup
public B group(EventLoopGroup group);
// 用来设置一个服务端的通道实现
public B channel(Class<? extends C> channelClass);
// 用来给 ServerChannel 添加配置
public <T> B option(ChannelOption<T> option, T value);
// 用来给接收到的通道添加配置
public <T> ServerBootstrap childOption(ChannelOption<T> option, T value);
// 用来设置业务处理类（自定义 handler）
public ServerBootstrap childHandler(ChannelHandler channelHandler);
// 用于服务器，用来设置占用的端口号
public ChannelFuture bind(int inetPort);
// 用于客户端，用来连接服务端
public ChannelFuture connect(String inetHost, int inetPort);

// 关闭通道
ChannelFuture close();
// 刷新
ChannelOutBoundInvoker flush();
// 将数据写到 ChannelPipeline 中当前 ChannelHandler 的下一个 ChannelHandler 开始处理（出站）
ChannelFuture writeAndFlush(Object msg);