• Netty入门案例 与 Netty异步模型

    Netty入门案例

    Netty 是由 JBOSS 提供的一个 Java 开源框架,所以在使用得时候首先得导入Netty的maven坐标

    1. <dependency>
    2.     <groupId>io.nettygroupId>
    3.     <artifactId>netty-allartifactId>
    4.     <version>4.1.42.Finalversion>
    5. dependency>

    1、Netty服务端编写

    服务端实现步骤:

    1. 创建bossGroup线程组: 处理网络事件--连接事件
    2. 创建workerGroup线程组: 处理网络事件--读写事件
    3. 创建服务端启动助手
    4. 设置bossGroup线程组和workerGroup线程组
    5. 设置服务端通道实现为NIO
    6. 参数设置
    7. 创建一个通道初始化对象
    8. 向pipeline中添加自定义业务处理handler
    9. 启动服务端并绑定端口,同时将异步改为同步
    10. 关闭通道和关闭连接池

    代码实现:

    1. package com.lagou.demo;
    2.  
    3. import io.netty.bootstrap.ServerBootstrap;
    4. import io.netty.channel.*;
    5. import io.netty.channel.nio.NioEventLoopGroup;
    6. import io.netty.channel.socket.SocketChannel;
    7. import io.netty.channel.socket.nio.NioServerSocketChannel;
    8.  
    9. /**
    10.  * Netty服务端
    11.  */
    12. public class NettyServer {
    13.     public static void main(String[] args) throws InterruptedException {
    14.         // 1. 创建bossGroup线程组: 处理网络事件--连接事件
    15.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    16.         // 2. 创建workerGroup线程组: 处理网络事件--读写事件 2*处理器线程数
    17.         EventLoopGroup workerGroup = new NioEventLoopGroup(2);
    18.         // 3. 创建服务端启动助手
    19.         ServerBootstrap serverBootstrap = new ServerBootstrap();
    20.         // 4. 设置bossGroup线程组和workerGroup线程组
    21.         serverBootstrap.group(bossGroup, workerGroup)
    22.                 .channel(NioServerSocketChannel.class)  // 5. 设置服务端通道实现为NIO
    23.                 .option(ChannelOption.SO_BACKLOG, 128)   // 6. 参数设置
    24.                 .childOption(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)   // 6. 参数设置
    25.                 .childHandler(new ChannelInitializer() {// 7. 创建一个通道初始化对象
    26.                     @Override
    27.                     protected void initChannel(SocketChannel ch) throws Exception {
    28.                         // 8. 向pipeline中添加自定义业务处理handler
    29.                         ch.pipeline().addLast(new NettyServerHandler());
    30.                     }
    31.                 });
    32.  
    33.         // 9. 启动服务端并绑定端口,同时将异步改为同步
    34.         // ChannelFuture future = serverBootstrap.bind(9999).sync();
    35.         ChannelFuture future = serverBootstrap.bind(9999);
    36.         future.addListener(new ChannelFutureListener() {
    37.             @Override
    38.             public void operationComplete(ChannelFuture future) throws Exception {
    39.                 if (future.isSuccess()) {
    40.                     System.out.println("端口绑定成功");
    41.                 } else {
    42.                     System.out.println("端口绑定失败");
    43.                 }
    44.             }
    45.         });
    46.         System.out.println("服务端启动成功");
    47.         // 10. 关闭通道(并不是真正意义上的关闭，而是监听通道关闭的状态)和关闭连接池
    48.         future.channel().closeFuture().sync();
    49.         bossGroup.shutdownGracefully();
    50.         workerGroup.shutdownGracefully();
    51.     }
    52. }

    自定义服务端handle

    1. package com.lagou.demo;
    2.  
    3. import io.netty.buffer.ByteBuf;
    4. import io.netty.buffer.Unpooled;
    5. import io.netty.channel.ChannelHandlerContext;
    6. import io.netty.channel.ChannelInboundHandler;
    7. import io.netty.util.CharsetUtil;
    8.  
    9. /**
    10.  * 自定义处理handler
    11.  */
    12. public class NettyServerHandler implements ChannelInboundHandler {
    13.  
    14.     /**
    15.      * 通道读取事件
    16.      *
    17.      * @param ctx
    18.      * @param msg
    19.      * @throws Exception
    20.      */
    21.     @Override
    22.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    23.         ByteBuf byteBuf = (ByteBuf) msg;
    24.         System.out.println("客户端发送过来的消息:" + byteBuf.toString(CharsetUtil.UTF_8));
    25.     }
    26.  
    27.     /**
    28.      * 通道读取完毕事件
    29.      *
    30.      * @param ctx
    31.      * @throws Exception
    32.      */
    33.     @Override
    34.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    35.         ctx.writeAndFlush(Unpooled.copiedBuffer("你好，我是Netty服务端", CharsetUtil.UTF_8)); // 消息出站
    36.  
    37.     }
    38.  
    39.     /**
    40.      * 通道异常事件
    41.      * @param ctx
    42.      * @param cause
    43.      * @throws Exception
    44.      */
    45.     @Override
    46.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    47.         cause.printStackTrace();
    48.         ctx.close();
    49.     }
    50.  
    51.     @Override
    52.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    53.  
    54.     }
    55.  
    56.     @Override
    57.     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    58.  
    59.     }
    60.  
    61.     @Override
    62.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
    63.  
    64.     }
    65.  
    66.     @Override
    67.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    68.  
    69.     }
    70.  
    71.     @Override
    72.     public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
    73.  
    74.     }
    75.  
    76.     @Override
    77.     public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
    78.  
    79.     }
    80.  
    81.     @Override
    82.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    83.  
    84.     }
    85.  
    86.     @Override
    87.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    88.  
    89.     }
    90. }

    2、Netty客户端编写

    客户端实现步骤:

    1. 创建线程组
    2. 创建客户端启动助手
    3. 设置线程组
    4. 设置客户端通道实现为NIO
    5. 创建一个通道初始化对象
    6. 向pipeline中添加自定义业务处理handler
    7. 启动客户端,等待连接服务端,同时将异步改为同步
    8. 关闭通道和关闭连接池

    代码实现:

    1. package com.lagou.demo;
    2.  
    3. import io.netty.bootstrap.Bootstrap;
    4. import io.netty.channel.ChannelFuture;
    5. import io.netty.channel.ChannelInitializer;
    6. import io.netty.channel.EventLoopGroup;
    7. import io.netty.channel.nio.NioEventLoopGroup;
    8. import io.netty.channel.socket.SocketChannel;
    9. import io.netty.channel.socket.nio.NioSocketChannel;
    10.  
    11. /**
    12.  * Netty客户端
    13.  */
    14. public class NettyClient {
    15.     public static void main(String[] args) throws InterruptedException {
    16.       //  1. 创建线程组
    17.         EventLoopGroup group = new NioEventLoopGroup();
    18.       //  2. 创建客户端启动助手
    19.         Bootstrap bootstrap = new Bootstrap();
    20.       //  3. 设置线程组
    21.         bootstrap.group(group)
    22.                 .channel(NioSocketChannel.class) //  4. 设置客户端通道实现为NIO
    23.                 .handler(new ChannelInitializer() { //  5. 创建一个通道初始化对象
    24.                     @Override
    25.                     protected void initChannel(SocketChannel ch) throws Exception {
    26.                         //  6. 向pipeline中添加自定义业务处理handler
    27.                         ch.pipeline().addLast(new NettyClientHandler());
    28.                     }
    29.                 });
    30.       //  7. 启动客户端,等待连接服务端,同时将异步改为同步
    31.         ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync();
    32.         //  8. 关闭通道和关闭连接池
    33.         channelFuture.channel().closeFuture().sync();
    34.         group.shutdownGracefully();
    35.     }
    36. }

    自定义客户端handle

    1. package com.lagou.demo;
    2.  
    3. import io.netty.buffer.ByteBuf;
    4. import io.netty.buffer.Unpooled;
    5. import io.netty.channel.ChannelFuture;
    6. import io.netty.channel.ChannelFutureListener;
    7. import io.netty.channel.ChannelHandlerContext;
    8. import io.netty.channel.ChannelInboundHandler;
    9. import io.netty.util.CharsetUtil;
    10.  
    11. /**
    12.  * 客户端处理类
    13.  */
    14. public class NettyClientHandler implements ChannelInboundHandler {
    15.     /**
    16.      * 通道就绪事件
    17.      *
    18.      * @param ctx
    19.      * @throws Exception
    20.      */
    21.     @Override
    22.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
    23.         // ctx.writeAndFlush(Unpooled.copiedBuffer("你好呀，我是Netty客户端", CharsetUtil.UTF_8));
    24.         ChannelFuture future = ctx.writeAndFlush(Unpooled.copiedBuffer("你好呀，我是Netty客户端", CharsetUtil.UTF_8));
    25.         future.addListener(new ChannelFutureListener() {
    26.             @Override
    27.             public void operationComplete(ChannelFuture future) throws Exception {
    28.                 if (future.isSuccess()) {
    29.                     System.out.println("数据发送成功");
    30.                 }else {
    31.                     System.out.println("数据发送失败");
    32.                 }
    33.             }
    34.         });
    35.     }
    36.  
    37.     /**
    38.      * 通道读就绪事件
    39.      *
    40.      * @param ctx
    41.      * @param msg
    42.      * @throws Exception
    43.      */
    44.     @Override
    45.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    46.         ByteBuf byteBuf = (ByteBuf) msg;
    47.         System.out.println("服务端发送的消息：" + byteBuf.toString(CharsetUtil.UTF_8));
    48.     }
    49.  
    50.     @Override
    51.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    52.  
    53.     }
    54.  
    55.     @Override
    56.     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    57.  
    58.     }
    59.  
    60.     @Override
    61.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    62.  
    63.     }
    64.  
    65.     @Override
    66.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    67.  
    68.     }
    69.  
    70.     @Override
    71.     public void userEventTriggered(ChannelHandlerContext ctx, Object msg) throws Exception {
    72.  
    73.     }
    74.  
    75.     @Override
    76.     public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
    77.  
    78.     }
    79.  
    80.     @Override
    81.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    82.  
    83.     }
    84.  
    85.     @Override
    86.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
    87.  
    88.     }
    89.  
    90.     @Override
    91.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable throwable) throws Exception {
    92.  
    93.     }
    94. }

    Netty异步模型

    1、基本介绍

            异步的概念和同步相对。当一个异步过程调用发出后，调用者不能立刻得到结果。实际处理这个调用的组件在完成后，通过状态、通知和回调来通知调用者。

            Netty 中的 I/O 操作是异步的，包括 Bind、Write、Connect 等操作会简单的返回一个ChannelFuture。调用者并不能立刻获得结果，而是通过 Future-Listener 机制，用户可以方便的主动获取或者通过通知机制获得IO 操作结果. Netty 的异步模型是建立在 future 和 callback 的之上的。callback 就是回调。重点说 Future，它的核心思想是：假设一个方法 fun，计算过程可能非常耗时，等待 fun 返回显然不合适。那么可以在调用 fun 的时候，立马返回一个 Future，后续可以通过 Future 去监控方法 fun 的处理过程(即 ： Future-Listener 机制) 

    2、Future 和Future-Listener

    Future

            表示异步的执行结果, 可以通过它提供的方法来检测执行是否完成，ChannelFuture 是他的一个子接口. ChannelFuture 是一个接口 ,可以添加监听器，当监听的事件发生时，就会通知到监听器

            当 Future 对象刚刚创建时，处于非完成状态，调用者可以通过返回的 ChannelFuture 来获取操作执行的状态， 注册监听函数来执行完成后的操作。

    常用方法有:

    方法名说明
    sync阻塞等待程序结果返回
    isDone来判断当前操作是否完成
    isSuccess来判断已完成的当前操作是否成功
    getCause来获取已完成的当前操作失败的原因
    isCancelled来判断已完成的当前操作是否被取消
    addListener来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知指定的监听器；如果Future 对象已完成，则通知指定的监听器

    Future-Listener 机制

            给Future添加监听器,监听操作结果

    代码实现:

    1. // 服务端部分代码
    2.  ChannelFuture future = serverBootstrap.bind(9999);
    3.         future.addListener(new ChannelFutureListener() {
    4.             @Override
    5.             public void operationComplete(ChannelFuture future) throws Exception {
    6.                 if (future.isSuccess()) {
    7.                     System.out.println("端口绑定成功");
    8.                 } else {
    9.                     System.out.println("端口绑定失败");
    10.                 }
    11.             }
    12.         });
    13.  
    14.  
    15. // 客户端部分代码
    16. ChannelFuture future = ctx.writeAndFlush(Unpooled.copiedBuffer("你好呀，我是Netty客户端", CharsetUtil.UTF_8));
    17.         future.addListener(new ChannelFutureListener() {
    18.             @Override
    19.             public void operationComplete(ChannelFuture future) throws Exception {
    20.                 if (future.isSuccess()) {
    21.                     System.out.println("数据发送成功");
    22.                 }else {
    23.                     System.out.println("数据发送失败");
    24.                 }
    25.             }
    26.         });
  • 相关阅读:
    Vue3+NodeJS 接入文心一言， 发布一个 VSCode 大模型问答插件
    云智慧联合北航提出智能运维（AIOps）大语言模型及评测基准
    当你使用ChatGPT时，选择合适的提示（prompt）是引导对话方向的关键
    [设计模式] 静态代理居然能解决这种问题，我惊讶了！
    如何理解Java中眼花缭乱的各种并发锁？
    C++中的多态和虚函数及多态原理
    Spring MVC 十一：中文乱码
    windows 深度学习环境部署
    使用 ES 实现疫情地图或者外卖点餐功能（含代码及数据）
    2024年特种设备作业人员考试题库及答案（流动式起重机Q2）
  • 原文地址：https://blog.csdn.net/weixin_52851967/article/details/126107209