devbox
笔触狂放9
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SpringBoot集成Netty_springboot集成netty服务端

作者：笔触狂放9 | 2024-05-15 10:39:55

赞

踩

springboot集成netty服务端

目录

一、pom依赖

二、配置yml文件

三、服务端

四、客户端

五、粘包和拆包问题

一、pom依赖

  1. <dependency>
  2.     <groupId>io.netty</groupId>
  3.     <artifactId>netty-all</artifactId>
  4.     <version>4.1.77.Final</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>cn.hutool</groupId>
  8.     <artifactId>hutool-all</artifactId>
  9.     <version>5.5.8</version>
  10. </dependency>

二、配置yml文件

  1. server:
  2.   port: 8001
  3.   servlet:
  4.     context-path: /netty
  5. netty:
  6.   url: 0.0.0.0  #0.0.0.0表示绑定任意ip
  7.   port: 20004

三、服务端

  1. package com.tlxy.lhn.controller.netty;
  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. public class NettyServer {
  10.     public static void main(String[] args) throws InterruptedException {
  11.         //创建两个线程组bossGroup和workerGroup，含有的子线程NioEventLoop的个数默认是CPU的两倍
  12.         //bossGroup只是处理连接请求，真正的和客户端业务处理，会交给workerGroup完成
  13.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  14.         EventLoopGroup workerGroup = new NioEventLoopGroup(1);
  15.  
  16.         try {
  17.             //创建服务器端的启动对象
  18.             ServerBootstrap bootstrap = new ServerBootstrap();
  19.             //使用链式编程来配置参数
  20.             bootstrap.group(bossGroup, workerGroup)//设置两个线程组
  21.                     .channel(NioServerSocketChannel.class)//使用NioServerSocketChannel作为服务器的通道实现
  22.                     //初始化服务器连接队列大小，服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接
  23.                     //多个客户端同时来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理
  24.                     .option(ChannelOption.SO_BACKLOG, 1024)
  25.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  26.                         @Override
  27.                         protected void initChannel(SocketChannel channel) throws Exception {
  28.                             //对workerGroup的SocketChannel设置处理器
  29.                             channel.pipeline().addLast(new NettyServerHandler());
  30.                         }
  31.                     });
  32.  
  33.             System.out.println("netty server start..");
  34.  
  35.             //绑定一个端口并且同步生成一个ChannelFuture异步对象，通过isDone()等方法可以判断异步事件的执行情况
  36.             //启动服务器（并绑定的端口），bind是异步操作，sync方法是等待异步操作执行完毕
  37.             ChannelFuture cf = bootstrap.bind(9000).sync();
  38.  
  39.             //给cf注册监听器，监听我们关心的事件
  40.             cf.addListener(new ChannelFutureListener() {
  41.                 @Override
  42.                 public void operationComplete(ChannelFuture channelFuture) throws Exception {
  43.                     if (cf.isSuccess()) {
  44.                         System.out.println("监听端口9000成功");
  45.                     } else {
  46.                         System.out.println("监听端口9000失败");
  47.                     }
  48.                 }
  49.             });
  50.             //等待服务端监听端口关闭，closeFuture是异步操作
  51.             //通过sync方法同步等待通道关闭处理完毕，这里会阻塞等待通道关闭完成，内部调用的是Object的wait()方法
  52.             cf.channel().closeFuture().sync();
  53.  
  54.         } finally {
  55.             bossGroup.shutdownGracefully();
  56.             workerGroup.shutdownGracefully();
  57.         }
  58.  
  59.  
  60.     }
  61. }
NettyServer类中的channel.pipeline().addLast(new NettyServerHandler());对应以下的处理器。
  1. package com.tlxy.lhn.controller.netty;
  2.  
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  8. import lombok.extern.slf4j.Slf4j;
  9.  
  10. @Slf4j
  11. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  12.  
  13.     @Override
  14.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  15.         ByteBuf buf = (ByteBuf) msg;
  16.         System.out.println("客户端发送消息是：" + buf.toString(CharsetUtil.UTF_8));
  17.         // 读取byteBuf
  18.         // 业务处理
  19.         // 回消息给客户端
  20.     }
  21.  
  22.     @Override
  23.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  24.         ByteBuf buf = Unpooled.copiedBuffer("HelloClient".getBytes(CharsetUtil.UTF_8));
  25.         ctx.writeAndFlush(buf);
  26.     }
  27.  
  28.     //只要Netty抛出错误就会执行,Netty断会开连接会抛出连接超时的错误
  29.     @Override
  30.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
  31.         log.info("关闭通道");
  32.         cause.printStackTrace();
  33.         ctx.close();
  34.     }
  35. }

四、客户端

  1. package com.tlxy.lhn.controller.netty;
  2.  
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.nio.NioEventLoopGroup;
  7. import io.netty.channel.socket.SocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  9.  
  10. public class NettyClient {
  11.     public static void main(String[] args) throws InterruptedException {
  12.         //客户端需要一个事件循环组
  13.         NioEventLoopGroup group = new NioEventLoopGroup();
  14.  
  15.         try {
  16.             //创建客户端启动对象
  17.             //注意客户端使用的不是SocketBootstrap而是Bootstrap
  18.             Bootstrap bootstrap = new Bootstrap();
  19.  
  20.             // 设置相关参数
  21.             bootstrap.group(group) //设置线程组
  22.                     .channel(NioSocketChannel.class)// 使用NioSocketChannel作为客户端的通道实现
  23.                     .handler(new ChannelInitializer<SocketChannel>() {
  24.                         @Override
  25.                         protected void initChannel(SocketChannel ch) throws Exception {
  26.                             ch.pipeline().addLast(new NettyClientHandler());
  27.                         }
  28.                     });
  29.  
  30.             System.out.println("netty client start..");
  31.             ChannelFuture cf = bootstrap.connect("127.0.0.1", 9000).sync();
  32.             cf.channel().closeFuture().sync();
  33.         }finally {
  34.             group.shutdownGracefully();
  35.         }
  36.  
  37.  
  38.     }
  39. }
NettyClient类中ch.pipeline().addLast(new NettyClientHandler());为处理器。
  1. package com.tlxy.lhn.controller.netty;
  2.  
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  8. import lombok.extern.slf4j.Slf4j;
  9.  
  10. @Slf4j
  11. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  12.  
  13.     /**
  14.      * 客户端连接标识
  15.      * @param ctx
  16.      * @throws Exception
  17.      */
  18.     @Override
  19.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  20.         ByteBuf buf = Unpooled.copiedBuffer("HelloServer".getBytes(CharsetUtil.UTF_8));
  21.         ctx.writeAndFlush(buf);
  22.     }
  23.  
  24.     //当通道建立后有事件时会触发，即服务端发送数据给客户端
  25.     @Override
  26.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  27.         ByteBuf buf = (ByteBuf) msg;
  28.         System.out.println("收到服务端的消息是：" + buf.toString(CharsetUtil.UTF_8));
  29.         System.out.println("服务端地址是：" + ctx.channel().remoteAddress());
  30.     }
  31.  
  32.     @Override
  33.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  34.         log.info("关闭通道");
  35.         cause.printStackTrace();
  36.         ctx.close();
  37.     }
  38.  
  39. }

五、粘包和拆包问题

客户端和服务端都是固定的框架，我们只需写处理器。

粘包和拆包问题，可以自己手写通过固定长度发送数据，或者使用Google的Protostuff。

  1. <dependency>
  2.     <groupId>com.dyuproject.protostuff</groupId>
  3.     <artifactId>protostuff-api</artifactId>
  4.     <version>1.0.8</version>
  5. </dependency>
  6. <dependency>
  7.     <groupId>com.dyuproject.protostuff</groupId>
  8.     <artifactId>protostuff-core</artifactId>
  9.     <version>1.0.8</version>
  10. </dependency>
  11. <dependency>
  12.     <groupId>com.dyuproject.protostuff</groupId>
  13.     <artifactId>protostuff-runtime</artifactId>
  14.     <version>1.0.8</version>
  15. </dependency>
  1. package com.tlxy.lhn.controller.netty;
  2.  
  3. import com.dyuproject.protostuff.LinkedBuffer;
  4. import com.dyuproject.protostuff.ProtostuffIOUtil;
  5. import com.dyuproject.protostuff.Schema;
  6. import com.dyuproject.protostuff.runtime.RuntimeSchema;
  7.  
  8. import java.util.Map;
  9. import java.util.concurrent.ConcurrentHashMap;
  10.  
  11. public class ProtostuffUtil {
  12.     private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<Class<?>, Schema<?>>();
  13.  
  14.     private static <T> Schema<T> getSchema(Class<T> clazz) {
  15.         @SuppressWarnings("unchecked")
  16.         Schema<T> schema = (Schema<T>) cachedSchema.get(clazz);
  17.         if (schema == null) {
  18.             schema = RuntimeSchema.getSchema(clazz);
  19.             if (schema != null) {
  20.                 cachedSchema.put(clazz, schema);
  21.             }
  22.         }
  23.         return schema;
  24.     }
  25.  
  26.     /**
  27.      * 序列化
  28.      *
  29.      * @param obj
  30.      * @return
  31.      */
  32.     public static <T> byte[] serializer(T obj) {
  33.         @SuppressWarnings("unchecked")
  34.         Class<T> clazz = (Class<T>) obj.getClass();
  35.         LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
  36.         try {
  37.             Schema<T> schema = getSchema(clazz);
  38.             return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
  39.         } catch (Exception e) {
  40.             throw new IllegalStateException(e.getMessage(), e);
  41.         } finally {
  42.             buffer.clear();
  43.         }
  44.     }
  45.  
  46.     /**
  47.      * 反序列化
  48.      *
  49.      * @param data
  50.      * @param clazz
  51.      * @return
  52.      */
  53.     public static <T> T deserializer(byte[] data, Class<T> clazz) {
  54.         try {
  55.             T obj = clazz.newInstance();
  56.             Schema<T> schema = getSchema(clazz);
  57.             ProtostuffIOUtil.mergeFrom(data, obj, schema);
  58.             return obj;
  59.         } catch (Exception e) {
  60.             throw new IllegalStateException(e.getMessage(), e);
  61.         }
  62.     }
  63.  
  64. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/572447
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号