个人随笔
目录
netty(三)、实现一个简单的在线聊天程序
2021-01-21 20:48:31

逻辑

客户端捕获键盘输入然后发送给服务端
服务器接收到消息，转发给各个客户端

服务端代码

  1. /**
  2.  * 服务端
  3.  * @author suibibk@qq.com
  4.  */
  5. public class NettyServer {
  6.     public static void main(String[] args) throws Exception{
  7.         //1、创建一个线程组，接收客户端连接
  8.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  9.         //2、创建一个线程组，处理网络操作
  10.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  11.         //3、创建服务端启动助手来配置参数
  12.         ServerBootstrap b = new ServerBootstrap();
  13.         b.group(bossGroup, workerGroup)//4、设置两个线程组
  14.          .channel(NioServerSocketChannel.class)//5、使用NioServerSocketChannel作为服务器端通道的实现
  15.          .option(ChannelOption.SO_BACKLOG, 128)//6、设置线程队列中等待连接的个数
  16.          .childOption(ChannelOption.SO_KEEPALIVE, true)//7、保持活动连接状态
  17.          .childHandler(new ChannelInitializer<SocketChannel>() {//8、创建一个通道初始化对象
  18.             @Override
  19.             protected void initChannel(SocketChannel sc) throws Exception {
  20.                 sc.pipeline().addLast(new StringEncoder());//对 String 对象自动编码,属于出站站处理器
  21.                 sc.pipeline().addLast(new StringDecoder());//把网络字节流自动解码为 String 对象，属于入站处理器
  22.                 //9、往pipeline链中添加自定义的handler
  23.                 sc.pipeline().addLast(new NettyServerHandler());
  24.             }
  25.          });
  26.         System.out.println("【服务器已经启动】");
  27.         ChannelFuture cf = b.bind(9999).sync();//10、绑定端口，bind方法是异步的，sync方法是同步阻塞的
  28.         //11、关闭通道，关闭线程组
  29.         cf.channel().closeFuture().sync();//关闭连接(异步非阻塞)
  30.         bossGroup.shutdownGracefully();
  31.         workerGroup.shutdownGracefully();
  32.     }
  33. }
  1. /**
  2.  * 服务端处理业务类
  3.  * @author suibibk@qq.com
  4.  *
  5.  */
  6. public class NettyServerHandler extends ChannelInboundHandlerAdapter{
  7.     //内部会用ConcurrentMap来维护，线程安全
  8.     private static ChannelGroup  channelGroup=new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  9.     private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  10.     /**
  11.      * channel处于就绪状态，客户端刚上线
  12.      */
  13.     @Override
  14.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  15.         Channel channel = ctx.channel();
  16.         String remoteAddress = channel.remoteAddress().toString();
  17.         //加入全局变量中
  18.         //channelGroup.writeAndFlush(Unpooled.copiedBuffer("【客户端】"+remoteAddress+"上线啦 "+format.format(new Date()),CharsetUtil.UTF_8));
  19.         channelGroup.writeAndFlush("【客户端】"+remoteAddress+"上线啦 "+format.format(new Date()));
  20.         channelGroup.add(channel);
  21.         //将当前channel加入到ChannelGroup
  22.         System.out.println("【客户端】"+remoteAddress+"上线啦");
  23.     }
  25.     /**
  26.      * channel离线
  27.      */
  28.     @Override
  29.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  30.         Channel channel = ctx.channel();
  31.         String remoteAddress = channel.remoteAddress().toString();
  32.         channelGroup.remove(channel);
  33.         channelGroup.writeAndFlush("【客户端】"+remoteAddress+"已下线 "+format.format(new Date()));
  34.         System.out.println("【客户端】"+remoteAddress+"已下线");
  35.     }
  36.     //读取数据事件
  37.     @Override
  38.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  39.         Channel channel = ctx.channel();
  40.         String remoteAddress = channel.remoteAddress().toString();
  41.         System.out.println("【客户端】"+remoteAddress+":"+msg);
  42.         channelGroup.forEach(ch -> {
  43.                 if(ch==channel) {
  44.                     ch.writeAndFlush("【自己】"+remoteAddress+":"+msg);
  45.                 }else {
  46.                     ch.writeAndFlush("【客户端】"+remoteAddress+":"+msg);
  47.                 }
  48.             }
  49.         );
  51.     }
  52.     //异常发生事件
  53.     @Override
  54.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  55.         //日志：远程主机强迫关闭了一个现有的连接。
  56.         //System.out.println(cause.getMessage());
  57.         ctx.close();
  58.     }
  59. }

客户端代码

  1. /**
  2.  * 网络客户端
  3.  * @author suibibk@qq.com
  4.  *
  5.  */
  6. public class NettyClient {
  7.     public static void main(String[] args)throws Exception {
  8.         //1、创建一个线程组
  9.         EventLoopGroup group = new NioEventLoopGroup();
  10.         //2、创建客户端启动助手，完成相关配置
  11.         Bootstrap b = new Bootstrap();
  12.         b.group(group)//3、设置线程组
  13.          .channel(NioSocketChannel.class)//4、设置客户端通道的实现类
  14.          .handler(new ChannelInitializer<SocketChannel>() {//创建一个初始化通道对象
  15.             @Override
  16.             protected void initChannel(SocketChannel sc) throws Exception {
  17.                 sc.pipeline().addLast(new StringEncoder());//对 String 对象自动编码,属于出站站处理器
  18.                 sc.pipeline().addLast(new StringDecoder());//把网络字节流自动解码为 String 对象，属于入站处理器
  19.                 //6、在pipline中添加自定义的handler
  20.                 sc.pipeline().addLast(new NettyClientHandler());
  21.             }
  22.          });
  23.         System.out.println("【客户端已启动】");
  24.         //7、启动客户端去连接服务器端 connect方法是异步的，sync方法是同步阻塞的
  25.         ChannelFuture cf =b.connect("127.0.0.1", 9999).sync();
  26.         System.out.println("---"+cf.channel().remoteAddress()+"------");
  27.         //循环监听用户键盘输入
  28.         Scanner scanner = new Scanner(System.in);
  29.         while(scanner.hasNextLine()) {
  30.             String msg = scanner.nextLine();
  31.             System.out.println("用户输入："+msg);
  32.             //通过channel发送到服务器端
  34.             cf.channel().writeAndFlush(msg);
  35.         }
  37.         //8、关闭连接（异步非阻塞）
  38.         cf.channel().closeFuture().sync();
  39.         System.out.print("Client is end.....");
  40.     }
  41. }
  1. /**
  2.  *     客户端业务处理类
  3.  * @author suibibk@qq.com
  4.  *
  5.  */
  6. public class NettyClientHandler extends ChannelInboundHandlerAdapter{
  7.     //读取数据事件
  8.     @Override
  9.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  10.        // ByteBuf buf = (ByteBuf)msg;
  11.         System.out.println(msg);
  12.     }
  13.     //异常发生事件
  14.     @Override
  15.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  16.         //日志：远程主机强迫关闭了一个现有的连接。
  17.         //System.out.println(cause.getMessage());
  18.         ctx.close();
  19.     }
  20. }

测试

启动服务器，然后启动三个客户端，输入消息回车，会看到如下日志

  1. 【服务器已经启动】
  2. 【客户端】/127.0.0.1:50576上线啦
  3. 【客户端】/127.0.0.1:50600上线啦
  4. 【客户端】/127.0.0.1:50631上线啦
  5. 【客户端】/127.0.0.1:50576:我是客户端1
  6. 【客户端】/127.0.0.1:50631:我是客户端2
  1. 【客户端已启动】
  2. ---/127.0.0.1:9999------
  3. 【客户端】/127.0.0.1:50600上线啦 2021-01-21 20:30:41
  4. 【客户端】/127.0.0.1:50631上线啦 2021-01-21 20:31:14
  5. 我是客户端1
  6. 用户输入：我是客户端1
  7. 【自己】/127.0.0.1:50576:我是客户端1
  8. 【客户端】/127.0.0.1:50631:我是客户端2
  1. 【客户端已启动】
  2. ---/127.0.0.1:9999------
  3. 【客户端】/127.0.0.1:50631上线啦 2021-01-21 20:31:14
  4. 【客户端】/127.0.0.1:50576:我是客户端1
  5. 【客户端】/127.0.0.1:50631:我是客户端2
  1. 【客户端已启动】
  2. ---/127.0.0.1:9999------
  3. 【客户端】/127.0.0.1:50576:我是客户端1
  4. 我是客户端2
  5. 用户输入：我是客户端2
  6. 【自己】/127.0.0.1:50631:我是客户端2

总结

其实netty开发框架模板是固定的，主要编写的类是两个处理引擎

  1. NettyServerHandler
  2. NettyClientHandler

期间可能会遇到一些错误，可以参考：netty(二)、channelGroup.writeAndFlush发送消息客户端接收不到的原因

 582

啊！这个可能是世界上最丑的留言输入框功能~


当然，也是最丑的留言列表

有疑问发邮件到 : suibibk@qq.com 侵权立删
Copyright : 个人随笔   备案号 : 粤ICP备18099399号-2