SpringBoot集成Netty服务器接收大量数据实例

依赖

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.68.Final</version>
</dependency>
 
创建Netty服务类

1. @Component
2. @RequiredArgsConstructor
3. public class NettyServerConfig {
5.     // 定义服务器端口号
6.     private final int port = 13030;
8.     // 注入JsonServerHandler处理器
9.     private final JsonServerHandler jsonServerHandler;
11.     // 初始化方法，在Bean创建时启动Netty服务器
12.     @PostConstruct
13.     public void startNetty() {
14.         new Thread(() -> {
15.             // 创建boss和worker两个线程组
16.             EventLoopGroup bossGroup = new NioEventLoopGroup();
17.             EventLoopGroup workerGroup = new NioEventLoopGroup();
18.             try {
19.                 // 创建ServerBootstrap实例以配置服务器
20.                 ServerBootstrap bootstrap = new ServerBootstrap();
21.                 bootstrap.group(bossGroup, workerGroup)
22.                         .channel(NioServerSocketChannel.class)  // 指定使用NIO传输Channel
23.                         .childHandler(new ChannelInitializer<SocketChannel>() {
24.                             @Override
25.                             protected void initChannel(SocketChannel ch) throws Exception {
26.                                 // 配置通道处理器，包括解码器、编码器和业务处理器
27.                                 ch.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
28.                                 ch.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
29.                                 ch.pipeline().addLast(jsonServerHandler);  // 使用注入的JsonServerHandler
30.                             }
31.                         })
32.                         .option(ChannelOption.SO_BACKLOG, 128)  // 设置队列大小
33.                         .childOption(ChannelOption.SO_KEEPALIVE, true);  // 保持连接
35.                 // 绑定端口并启动服务器
36.                 ChannelFuture f = bootstrap.bind(port).sync();
37.                 // 等待服务器通道关闭
38.                 f.channel().closeFuture().sync();
39.             } catch (InterruptedException e) {
40.                 e.printStackTrace();
41.             } finally {
42.                 // 优雅地关闭线程组
43.                 workerGroup.shutdownGracefully();
44.                 bossGroup.shutdownGracefully();
45.             }
46.         }).start();
47.     }
48. }

复制代码

创建EchoServerHandler 类接收并存储数据

1. @Slf4j
2. @ChannelHandler.Sharable
3. @Component
4. public class JsonServerHandler extends SimpleChannelInboundHandler<String> {
5.     @Autowired
6.     private ReceptionMapper receptionMapper;
8.     private final ObjectMapper objectMapper = new ObjectMapper();
10.     // 处理接收到的消息
11.     @Override
12.     @Transactional
13.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
14.         try {
15.             // 反序列化接收到的JSON消息
16.             ReceptionData receptionData = null;
17.             try {
18.                 receptionData = objectMapper.readValue(msg, ReceptionData.class);
19.             } catch (JsonProcessingException e) {
20.                 log.error("解码数据错误", e);
21.             }
22.             //存入数据库
23.             receptionMapper.saveData(receptionData );
25.         } catch (Exception e) {
26.             log.error("接收数据错误", e);
27.         }
28.     }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。