【Netty】Springboot整合Netty

/**
 * 服务端
 * 1.创建一个ServerBootstrap的实例引导和绑定服务器。
 * 2.创建并分配一个NioEventLoopGroup实例以进行事件的处理，比如接受连接以及读写数据。
 * 3.指定服务器绑定的本地的InetSocketAddress。
 * 4.使用一个EchoServerHandler的实例初始化每一个新的Channel。
 * 5.调用ServerBootstrap.bind()方法以绑定服务器。
 */
@Slf4j
@Component
public class EchoServer {
 /**
     * NioEventLoop并不是一个纯粹的I/O线程，它除了负责I/O的读写之外
     * 创建了两个NioEventLoopGroup，
     * 它们实际是两个独立的Reactor线程池。
     * 一个用于接收客户端的TCP连接，
     * 另一个用于处理I/O相关的读写操作，或者执行系统Task、定时任务Task等。
     */
 private final EventLoopGroup bossGroup = new NioEventLoopGroup();
 private final EventLoopGroup workerGroup = new NioEventLoopGroup();
 private Channel channel;
 /**
     * 启动服务
     * @param hostname
     * @param port
     * @return
     * @throws Exception
     */
 public ChannelFuture  start(String hostname,int port) throws Exception {
 final EchoServerHandler serverHandler = new EchoServerHandler();
 ChannelFuture f = null;
 try {
 //ServerBootstrap负责初始化netty服务器，并且开始监听端口的socket请求
 ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
 .channel(NioServerSocketChannel.class)
 .localAddress(new InetSocketAddress(hostname,port))
 .childHandler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel socketChannel) throws Exception {
//                            为监听客户端read/write事件的Channel添加用户自定义的ChannelHandler
                            socketChannel.pipeline().addLast(serverHandler);
 }
 });
            f = b.bind().sync();
            channel = f.channel();
            log.info("======EchoServer启动成功!!!=========");
 } catch (Exception e) {
            e.printStackTrace();
 } finally {
 if (f != null && f.isSuccess()) {
                log.info("Netty server listening " + hostname + " on port " + port + " and ready for connections...");
 } else {
                log.error("Netty server start up Error!");
 }
 }
 return f;
 }
 /**
     * 停止服务
     */
 public void destroy() {
        log.info("Shutdown Netty Server...");
 if(channel != null) { channel.close();}
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
        log.info("Shutdown Netty Server Success!");
 }
}

/***
 * 服务端自定义业务处理handler
 */
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
 /**
     * 对每一个传入的消息都要调用；
     * @param ctx
     * @param msg
     * @throws Exception
     */
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 ByteBuf in = (ByteBuf) msg;
 System.out.println("server received: "+in.toString(CharsetUtil.UTF_8));
        ctx.write(in);
 }
 /**
     * 通知ChannelInboundHandler最后一次对channelRead()的调用时当前批量读取中的的最后一条消息。
     * @param ctx
     * @throws Exception
     */
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
 }
 /**
     * 在读取操作期间，有异常抛出时会调用。
     * @param ctx
     * @param cause
     * @throws Exception
     */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
 }
}

@SpringBootApplication
public class SpringNettyApplication implements CommandLineRunner {
 @Value("${netty.port}")
 private int port;
 @Value("${netty.url}")
 private String url;
 @Autowired
 private EchoServer echoServer;
 public static void main(String[] args) {
 SpringApplication.run(SpringNettyApplication.class, args);
 }
 @Override
 public void run(String... args) throws Exception {
 ChannelFuture future = echoServer.start(url,port);
 Runtime.getRuntime().addShutdownHook(new Thread(){
 @Override
 public void run() {
                echoServer.destroy();
 }
 });
 //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程
        future.channel().closeFuture().syncUninterruptibly();
 }
}

/**
 * 客户端
 * 1.为初始化客户端，创建一个Bootstrap实例
 * 2.为进行事件处理分配了一个NioEventLoopGroup实例，其中事件处理包括创建新的连接以及处理入站和出站数据；
 * 3.当连接被建立时，一个EchoClientHandler实例会被安装到（该Channel的一个ChannelPipeline中；
 * 4.在一切都设置完成后，调用Bootstrap.connect()方法连接到远程节点。
 */
public class EchoClient {
 private final String host;
 private final int port;
 public EchoClient(String host, int port) {
 this.host = host;
 this.port = port;
 }
 /**
     * 运行流程：
     * @param args
     * @throws Exception
     */
 public static void main(String[] args) throws Exception {
 new EchoClient("127.0.0.1",10010).start();
 }
 private void start() throws Exception {
 /**
         * Netty用于接收客户端请求的线程池职责如下。
         * （1）接收客户端TCP连接，初始化Channel参数；
         * （2）将链路状态变更事件通知给ChannelPipeline
         */
 EventLoopGroup group = new NioEventLoopGroup();
 try {
 Bootstrap b = new Bootstrap();
            b.group(group)
 .channel(NioSocketChannel.class)
 .remoteAddress(new InetSocketAddress(host,port))
 .handler(new ChannelInitializer<SocketChannel>() {
 @Override
 protected void initChannel(SocketChannel socketChannel) throws Exception {
                            socketChannel.pipeline().addLast(new EchoClientHandler());
 }
 });
 //绑定端口
 ChannelFuture f = b.connect().sync();
            f.channel().closeFuture().sync();
 } catch (Exception e) {
 group.shutdownGracefully().sync();
 }
 }
}

/**
 * 客户端处理类
 */
public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
 /**
     * 在到服务器的连接已经建立之后将被调用
     * @param ctx
     * @throws Exception
     */
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks !", CharsetUtil.UTF_8));
 }
 /**
     * 当从服务器接收到一个消息时被调用
     * @param channelHandlerContext
     * @param byteBuf
     * @throws Exception
     */
 @Override
 protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
 System.out.println("Client received: "+ byteBuf.toString(CharsetUtil.UTF_8));
 }
 /**
     * 在处理过程中引发异常时被调用
     * @param ctx
     * @param cause
     * @throws Exception
     */
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
 }
}