登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›springboot的netty代码实操

springboot的netty代码实操

上官银柳 2025-6-9 08:59:34
参考：https://www.cnblogs.com/mc-74120/p/13622008.html
pom文件
  1. <dependency>
  2.     <groupId>io.netty</groupId>
  3.     netty-all</artifactId>
  4. </dependency>
复制代码
启动类
  1. @EnableFeignClients
  2. @EnableDiscoveryClient
  3. @EnableScheduling
  4. @SpringBootApplication
  5. @EnableAsync
  6. public class ChimetaCoreApplication  implements CommandLineRunner{
  7.    
  8.     @Autowired
  9.     private NettyServerListener nettyServerListener;
  10.    
  11.     public static void main(String[] args) {
  12.         SpringApplication.run(ChimetaCoreApplication.class, args);
  13.     }
  14.    
  15.     @Override
  16.     public void run(String... args) throws Exception {
  17.       
  18.        nettyServerListener.start();
  19.     }
  20. }
复制代码
服务端代码的listener
  1. package com.chimeta.netty;
  3. import javax.annotation.PreDestroy;
  4. import javax.annotation.Resource;
  6. import org.springframework.beans.factory.annotation.Value;
  7. import org.springframework.stereotype.Component;
  9. import com.chimeta.netty.protobuf.ImProto;
  11. import io.netty.bootstrap.ServerBootstrap;
  12. import io.netty.channel.ChannelFuture;
  13. import io.netty.channel.ChannelInitializer;
  14. import io.netty.channel.ChannelOption;
  15. import io.netty.channel.EventLoopGroup;
  16. import io.netty.channel.nio.NioEventLoopGroup;
  17. import io.netty.channel.socket.SocketChannel;
  18. import io.netty.channel.socket.nio.NioServerSocketChannel;
  19. import io.netty.handler.codec.protobuf.ProtobufDecoder;
  20. import io.netty.handler.codec.protobuf.ProtobufEncoder;
  21. import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
  22. import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
  23. import io.netty.handler.logging.LogLevel;
  24. import io.netty.handler.logging.LoggingHandler;
  25. import io.netty.handler.timeout.IdleStateHandler;
  26. import lombok.extern.slf4j.Slf4j;
  28. /**
  29. * 服务启动监听器
  30. *
  31. * @author mwan
  32. */
  33. @Component
  34. @Slf4j
  35. public class NettyServerListener {
  36.     /**
  37.      * 创建bootstrap
  38.      */
  39.     ServerBootstrap serverBootstrap = new ServerBootstrap();
  40.     /**
  41.      * BOSS
  42.      */
  43.     EventLoopGroup boss = new NioEventLoopGroup();
  44.     /**
  45.      * Worker
  46.      */
  47.     EventLoopGroup work = new NioEventLoopGroup();
  48.     /**
  49.      * 通道适配器
  50.      */
  51.     @Resource
  52.     private ServerChannelHandlerAdapter channelHandlerAdapter;
  53.     /**
  54.      * 从配置中心获取NETTY服务器配置
  55.      */
  56.     @Value("${server.netty.port:10001}")
  57.     private int NETTY_PORT;
  58.    
  59.     @Value("${server.netty.maxthreads:5000}")
  60.     private int MAX_THREADS;
  62.     /**
  63.      * 关闭服务器方法
  64.      */
  65.     @PreDestroy
  66.     public void close() {
  67.         log.info("关闭服务器....");
  68.         //优雅退出
  69.         boss.shutdownGracefully();
  70.         work.shutdownGracefully();
  71.     }
  73.     /**
  74.      * 开启及服务线程
  75.      */
  76.     public void start() {
  77.         serverBootstrap.group(boss, work)
  78.                 .channel(NioServerSocketChannel.class)
  79.                 .option(ChannelOption.SO_BACKLOG, MAX_THREADS) //最大客户端连接数为1024  
  80.                 .handler(new LoggingHandler(LogLevel.INFO)).childOption(ChannelOption.SO_KEEPALIVE, true); ;
  81.         try {
  82.             //设置事件处理
  83.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  84.                 @Override
  85.                 protected void initChannel(SocketChannel ch) throws Exception {
  86.                     // 下面的每一个addLast都有自己的含义，需要每个都过一下
  87.                     ch.pipeline().addLast(new IdleStateHandler(18,0,0));
  88.                     ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
  89.                     //ch.pipeline().addLast(new CustomProtobufInt32FrameDecoder());
  90.                     ch.pipeline().addLast(new ProtobufDecoder(ImProto.ImMsg.getDefaultInstance()));
  91.                     ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
  92.                     //ch.pipeline().addLast(new CustomProtobufInt32LengthFieldPrepender());
  93.                     ch.pipeline().addLast(new ProtobufEncoder());
  94.                     // 业务处理
  95.                     ch.pipeline().addLast(channelHandlerAdapter);
  96.                 }
  97.             });
  98.             log.info("netty服务器在[{}]端口启动监听", NETTY_PORT);
  99.             ChannelFuture f = serverBootstrap.bind(NETTY_PORT).sync();
  100.             f.channel().closeFuture().sync();
  101.         } catch (InterruptedException e) {
  102.             log.error("[出现异常] 释放资源", e);
  103.             boss.shutdownGracefully();
  104.             work.shutdownGracefully();
  105.             log.info("服务已关闭！");
  106.         }
  107.     }
  108. }
复制代码
 ServerChannelHandlerAdapter处理类
  1. package com.chimeta.netty;
  3. import org.apache.commons.lang3.StringUtils;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.stereotype.Component;
  7. import com.chimeta.netty.model.SessionCloseReason;
  8. import com.chimeta.netty.protobuf.ImProto.ImMsg;
  9. import com.chimeta.netty.util.ChannelUtils;
  10. import com.google.protobuf.InvalidProtocolBufferException;
  11. import com.google.protobuf.util.JsonFormat;
  12. import io.netty.channel.ChannelHandler.Sharable;
  13. import io.netty.handler.timeout.IdleState;
  14. import io.netty.handler.timeout.IdleStateEvent;
  15. import io.netty.channel.Channel;
  16. import io.netty.channel.ChannelHandlerContext;
  17. import io.netty.channel.ChannelInboundHandlerAdapter;
  18. import lombok.extern.slf4j.Slf4j;
  20. /**
  21. * 通信服务处理器
  22. */
  23. @Component
  24. @Sharable
  25. @Slf4j
  26. public class ServerChannelHandlerAdapter extends ChannelInboundHandlerAdapter {
  27.     /**
  28.      * 注入请求分排器
  29.      */
  30.     @Autowired
  31.     private MessageDispatcher messageDispatcher;
  32.    
  33.     @Autowired
  34.     private DeviceSessionManager sessionManager;
  36.     /** 用来记录当前在线连接数。应该把它设计成线程安全的。  */
  37.     //private AtomicInteger sessionCount = new AtomicInteger(0);
  38.    
  39.     @Override  
  40.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {  
  41.         super.handlerAdded(ctx);
  42.         
  43.         if (!ChannelUtils.addChannelSession(ctx.channel(), new IoSession(ctx.channel()))) {
  44.           ctx.channel().close();
  45.           log.error("Duplicate session,IP=[{}]",ChannelUtils.getRemoteIp(ctx.channel()));
  46.        }     
  48.         //String server_ip = NetworkUtils.getRealIp();//获得本机IP
  49.         // 缓存计数器加1
  50.     }  
  51.       
  52.     @Override  
  53.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {  
  54.         super.handlerRemoved(ctx);
  56.         // 缓存计数器减1
  57.         //String server_ip = NetworkUtils.getRealIp();//获得本机IP
  58.         log.info(ctx.channel().id()+"离开了");  
  59.     }
  60.    
  61.     @Override
  62.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  63.         
  64.         ImMsg gameMessage = (ImMsg)msg;
  65.         final Channel channel = ctx.channel();
  66.        IoSession session = ChannelUtils.getSessionBy(channel);
  67.        if(session.isHeartbeated()) {
  68.           session.setHeartbeated(false);
  69.        }
  70.       
  71.        String deviceCode="";
  72.        if(session.getDevice() != null && StringUtils.isNotBlank(session.getDevice().getDeviceCode())) {
  73.           deviceCode = session.getDevice().getDeviceCode();
  74.        }
  75. //     if(!MessagingConst.TYPE_UPOS_REQUEST.equals(gameMessage.getMsg().getTypeUrl())) {
  76.           try {
  77.              log.info("Inbound message is :" + JsonFormat.printer().usingTypeRegistry(DeviceSessionManager.typeRegistry).print(gameMessage.toBuilder())
  78.                    + ", from device " + deviceCode);
  79.           } catch (InvalidProtocolBufferException e) {
  80.              log.info("Inbound message is :" + gameMessage.toString());
  81.           }
  82. //     }
  83.       
  84.        messageDispatcher.dispatch(gameMessage, session);
  85.     }
  86.      
  87.     @Override
  88.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {  
  89.         ctx.flush();  
  90.     }
  91.    
  92.     @Override  
  93.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)  
  94.             throws Exception {  
  95.         
  96.         log.error("通信发生异常：", cause);
  97.         ctx.close();   
  98.     }
  99.    
  100.     /**
  101.      * 一段时间未进行读写操作 回调
  102.      */
  103.     @Override
  104.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  105.         /*心跳处理*/
  106.         if (evt instanceof IdleStateEvent) {
  107.             IdleStateEvent event = (IdleStateEvent) evt;
  108.             if (event.state() == IdleState.READER_IDLE) {
  109.                 /*读超时*/
  110.                 log.info("READER_IDLE read overtime，close session");
  111.                 final Channel channel = ctx.channel();
  112.                IoSession session = ChannelUtils.getSessionBy(channel);
  113.                
  114.              /*
  115.               * if(messageDispatcher.sendHeartbeat(session) == false) { //如果心跳检测失败，则连接异常，主动断开
  116.               * session.setSessionCloseReason(SessionCloseReason.OVER_TIME); ctx.close(); };
  117.               */
  118.                
  119.                session.setSessionCloseReason(SessionCloseReason.OVER_TIME);
  120.               ctx.close();
  121.                
  122.             } else if (event.state() == IdleState.WRITER_IDLE) {
  123.                 /*写超时*/   
  124.                 log.info("WRITER_IDLE 写超时");
  125.             } else if (event.state() == IdleState.ALL_IDLE) {
  126.                 /*总超时*/
  127.                 log.info("ALL_IDLE 总超时");
  128.             }
  129.         }
  130.     }
  132.     @Override
  133.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  135.         sessionManager.unregisterUserContext(ctx.channel());
  136.         log.info(ctx.channel().id() + "已掉线！");
  137.         // 这里加入玩家的掉线处理
  138.         ctx.close();
  140.     }
  142. }
复制代码
MessageDispatcher分派各个处理器
  1. package com.chimeta.netty;
  3. import com.chimeta.netty.service.TerminalService;
  4. import org.springframework.beans.factory.annotation.Autowired;
  5. import org.springframework.scheduling.annotation.Async;
  6. import org.springframework.stereotype.Component;
  8. import com.chimeta.netty.constant.MessagingConst;
  9. import com.chimeta.netty.model.SessionCloseReason;
  10. import com.chimeta.netty.protobuf.ImProto.ImMsg;
  11. import com.chimeta.netty.service.LoginService;
  12. import com.chimeta.netty.util.MessageBuilder;
  13. import com.google.protobuf.InvalidProtocolBufferException;
  14. import lombok.extern.slf4j.Slf4j;
  16. import javax.annotation.Resource;
  18. /**
  19. * 请求分排器
  20. */
  21. @Component
  22. @Slf4j
  23. public class MessageDispatcher{
  24.    
  25.     @Autowired
  26.     private LoginService loginService;
  28.     @Resource
  29.     private TerminalService terminalService;
  30.    
  31.     /**
  32.      * 消息分发处理
  33.      *
  34.      * @param gameMsg
  35.      * @throws InvalidProtocolBufferException
  36.      */
  37.     @Async
  38.     public void dispatch(ImMsg imMsg, IoSession currSession) throws InvalidProtocolBufferException {
  40.         if(imMsg.getId() < 0) {
  41.            currSession.sendMessage(MessageBuilder.buildErrorResponse(imMsg, MessagingConst.RESPONSE_ERR_CODE_400, "Invalid message！"));
  42.            return;
  43.         }
  44.         //log.info("接收到的消息TypeUrl是: "+imMsg.getMsg().getTypeUrl());
  45.         switch(imMsg.getMsg().getTypeUrl()) {
  46.         
  47.             case MessagingConst.TYPE_ONLINE_REQUEST:
  48.                // 处理设备上线请求
  49.                loginService.doLogin(imMsg, currSession);
  50.                break;
  51.             case MessagingConst.TYPE_USER_LOGON_REQUEST:
  52.                // 处理请求
  53.                loginService.doUserLogon(imMsg, currSession);
  54.                break;
  55.             case MessagingConst.TYPE_USER_LOGOFF_REQUEST:
  56.                // 处理请求
  57.                loginService.doUserLogoff(imMsg, currSession);
  58.                break;
  59.           case MessagingConst.TYPE_TERMINAL_STATE_REQUEST:
  60.                 // 我写的
  61.              terminalService.multiInsert(imMsg, currSession);
  62.              break;
  63.             default:
  64.                if(currSession != null) {
  65.                   // 返回客户端发来的心跳消息
  66.                   responseHeartbeat(imMsg, currSession);
  67.                }
  68.                break;
  69.         }
  70.     }
  71.    
  72.     /**
  73.      * 发送心跳包消息
  74.      * @param gameMsg
  75.      * @param currSession
  76.      * @return
  77.      */
  78.     public boolean sendHeartbeat(IoSession currSession) {
  79.       
  80.        try {
  81.           if(currSession.isHeartbeated()) {
  82.              return false;
  83.           }
  84.           ImMsg.Builder imMsgBuilder = ImMsg.newBuilder();
  85.          
  86.           currSession.sendMessage(imMsgBuilder.build());
  87.          
  88.           currSession.setHeartbeated(true);
  89.          
  90.           return true;
  91.        }catch(Exception e) {
  92.           log.error("主动发送心跳包时发生异常：", e);
  93.           currSession.close(SessionCloseReason.EXCEPTION);
  94.           return false;
  95.        }
  96.       
  97.     }
  98.     /**
  99.      * 返回客户端发来的心跳包消息
  100.      * @param imMsg
  101.      * @param currSession
  102.      */
  103.     private void responseHeartbeat(ImMsg imMsg,IoSession currSession) {
  104.        ImMsg.Builder imMsgBuilder = ImMsg.newBuilder();
  105.       
  106.        currSession.sendMessage(imMsgBuilder.build());
  107.     }
  108.    
  109. }
复制代码
最后到service业务处理TerminalService
  1. package com.chimeta.netty.service;
  3. import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
  4. import com.chimeta.common.entity.terminal.TerminalStateMonitorDO;
  5. import com.chimeta.netty.IoSession;
  6. import com.chimeta.netty.constant.MessagingConst;
  7. import com.chimeta.netty.model.DeviceInfo;
  8. import com.chimeta.netty.protobuf.ImProto;
  9. import com.chimeta.netty.util.MessageBuilder;
  10. import com.chimeta.terminal.mapper.TerminalStateMonitorMapper;
  11. import lombok.extern.slf4j.Slf4j;
  12. import org.springframework.stereotype.Service;
  13. import org.springframework.transaction.annotation.Transactional;
  14. import org.springframework.util.CollectionUtils;
  16. import java.math.BigDecimal;
  17. import java.util.ArrayList;
  18. import java.util.List;
  20. /**
  21. * 盒子设备相关的实现类
  22. */
  23. @Service
  24. @Slf4j
  25. public class TerminalService extends ServiceImpl<TerminalStateMonitorMapper, TerminalStateMonitorDO> {
  27.     @Transactional(rollbackFor = Exception.class)
  28.     public void multiInsert(ImProto.ImMsg imMsg, IoSession currSession){
  29.         DeviceInfo deviceInfo = currSession.getDevice();
  30.         if(deviceInfo == null) {
  31.             currSession.sendMessage(MessageBuilder.buildErrorResponse(imMsg, MessagingConst.RESPONSE_ERR_CODE_400, "device not online!"));
  32.             return;
  33.         }
  34.         try {
  35.             ImProto.TerminalStateList terminalStateList = imMsg.getMsg().unpack(ImProto.TerminalStateList.class);
  36.             log.info("TerminalService multiInsert TerminalStateList:{}", terminalStateList);
  37.             List<ImProto.TerminalState> requestTerminalStateList = terminalStateList.getTerminalStateList();
  39.             if (!CollectionUtils.isEmpty(requestTerminalStateList)){
  40.                 List<TerminalStateMonitorDO> tmplist = new ArrayList<>();
  41.                 for (ImProto.TerminalState requestTerminalState : requestTerminalStateList){
  42.                     TerminalStateMonitorDO terminalStateMonitorDO = new TerminalStateMonitorDO();
  43.                     terminalStateMonitorDO.setBatteryLevel(requestTerminalState.getBatteryLevel());
  44.                     terminalStateMonitorDO.setChargingState(requestTerminalState.getChargingState());
  45.                     terminalStateMonitorDO.setTemperature(BigDecimal.valueOf(requestTerminalState.getTemperature()));
  46.                     terminalStateMonitorDO.setUniqueCode(deviceInfo.getDeviceCode());
  47.                     terminalStateMonitorDO.setStateTime(requestTerminalState.getStateTime());
  48.                     tmplist.add(terminalStateMonitorDO);
  49.                 }
  50.                 this.saveBatch(tmplist);
  51.             }
  52.         } catch (Exception e) {
  53.             log.error("TerminalService multiInsert error:{}", e);
  54.         }
  56.     }
  58. }
复制代码
至此，服务端的处理逻辑写完，然后比较费时间的是自己写client的请求，终于经过两三天时间总结好了，写了个test类，如下
  1. package com.chimeta.core;
  3. import com.chimeta.netty.protobuf.ImProto;
  4. import com.google.protobuf.Any;
  5. import io.netty.bootstrap.Bootstrap;
  6. import io.netty.buffer.Unpooled;
  7. import io.netty.channel.Channel;
  8. import io.netty.channel.ChannelInitializer;
  9. import io.netty.channel.nio.NioEventLoopGroup;
  10. import io.netty.channel.socket.nio.NioSocketChannel;
  11. import io.netty.handler.codec.protobuf.ProtobufEncoder;
  12. import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
  13. import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
  14. import lombok.extern.slf4j.Slf4j;
  15. import org.junit.jupiter.api.Test;
  16. import org.junit.runner.RunWith;
  17. import org.mockito.junit.MockitoJUnitRunner;
  20. @Slf4j
  21. @RunWith(MockitoJUnitRunner.class)
  22. class NettyTerminalTest {
  25.     @Test
  26.     public void tryTest()  throws InterruptedException {
  28.         ImProto.TerminalStateList terminalstateList = ImProto.TerminalStateList.newBuilder().build();
  29.         for (int i = 0; i < 3; i++) {
  30.             ImProto.TerminalState build = ImProto.TerminalState.newBuilder()
  31.                     .setBatteryLevel(i)
  32.                     .setChargingState(i * 11)
  33.                     .setTemperature(i * 11.1)
  34.                     .setStateTime(i * 111)
  35.                     .build();
  36.             terminalstateList = terminalstateList.toBuilder().addTerminalState(build).build();
  37.         }
  39.         ImProto.ImMsg imMsg = ImProto.ImMsg.newBuilder().setId(66).setMsg(Any.pack(terminalstateList)).build();
  41.         Channel channel = new Bootstrap()
  42.                 .group(new NioEventLoopGroup(1))
  43.                 .handler(new ChannelInitializer<NioSocketChannel>() {
  44.                     @Override
  45.                     protected void initChannel(NioSocketChannel ch) throws Exception {
  46.                         System.out.println("初始化连接...");
  47.                         ch.pipeline().addLast("encode", new ProtobufEncoder())
  48.                                 .addLast(new ProtobufVarint32FrameDecoder()).addLast(new ProtobufVarint32LengthFieldPrepender());
  49.                     }
  50.                 })
  51.                 .channel(NioSocketChannel.class).connect("192.168.123.123", 10001)
  52.                 .sync()
  53.                 .channel();
  55. //        channel.pipeline().addLast(new StringEncoder()).writeAndFlush(ByteBufAllocator.DEFAULT.buffer().writeBytes(imMsg.toByteArray()));
  56.         channel.pipeline().writeAndFlush(Unpooled.copiedBuffer(imMsg.toByteArray()));
  57.         System.out.println("over！");
  58.     }
  60. }
复制代码
 好了，记录下，以后就不会忘记了
 

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

使用道具 举报
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
上官银柳
2025-6-9 08:59:34

0

粉丝关注

16

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫