Spring Boot整合Netty（二）：注入Service

1、添加依赖

 1.          <!--netty-->
    
2.          <dependency>
    
3.              <groupId>io.netty</groupId>
    
4.              <artifactId>netty-all</artifactId>
    
5.              <version>4.1.24.Final</version>
    
6.              <scope>compile</scope>
    
7.          </dependency>

2、创建TcpServer

 2.  @Slf4j
    
3.  @Component
    
4.  @Order(value = 1)
    
5.  public class TcpServer implements CommandLineRunner {
    

7.      @Value("${netty.ip}")
    
8.      private String ip;
    

10.     @Value("${netty.port}")
    
11.     private int tcpPort;
    

13.     @Autowired
    
14.     private ProduceOrderService produceOrderService;
    

16.     @Autowired
    
17.     private TestRecordTopService testRecordTopService;
    

19.     @Autowired
    
20.     private TestRecordAllService testRecordAllService;
    

22.     @Autowired
    
23.     private ProduceOrderLogService produceOrderLogService;
    

25.     @Async
    
26.     @Override
    
27.     public void run(String... args) {
    
28.         start(ip, tcpPort);
    
29.     }
    

31.     public void start(String ip, int port) {
    
32.         // master reactor
    
33.         EventLoopGroup bossGroup = new NioEventLoopGroup();
    
34.         // sub reactor
    
35.         EventLoopGroup workGroup = new NioEventLoopGroup();
    

37.         ChannelFuture f;
    
38.         try {
    
39.             // start
    
40.             ServerBootstrap bootstrap = new ServerBootstrap();
    
41.             // install NioEventLoopGroup
    
42.             bootstrap.group(bossGroup, workGroup)
    
43.                     // set channel type NIO
    
44.                     .channel(NioServerSocketChannel.class)
    
45.                     // set connect params
    
46.                     .option(ChannelOption.SO_BACKLOG, 1024)
    
47.                     .childOption(ChannelOption.SO_KEEPALIVE, true)
    
48.                     .childOption(ChannelOption.TCP_NODELAY, true)
    
49.                     // set in/out event handler
    
50.                     .childHandler(new ChannelInitializer<NioSocketChannel>() {
    
51.                         @Override
    
52.                         protected void initChannel(NioSocketChannel ch) {
    
53.                             // 添加分隔符解码器
    
54.                             ch.pipeline().addFirst(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Unpooled.copiedBuffer("&&".getBytes())));
    
55.                             ch.pipeline().addLast(new TcpMsgOutHandler());
    
56.                             ch.pipeline().addLast(new TcpActiveHandler());
    
57.                             ch.pipeline().addLast(new TcpMsgInHandler(produceOrderService, testRecordTopService, testRecordAllService, produceOrderLogService));
    
58.                         }
    
59.                     });
    
60.             // bind port
    
61.             f = bootstrap.bind(ip, port).sync();
    
62.             if (f.isSuccess()) {
    
63.                 log.info("start tcp server success: {}", port);
    
64.             } else {
    
65.                 log.warn("start tcp server failed: {}", f.cause().getMessage());
    
66.             }
    
67.             f.channel().closeFuture().sync();
    
68.         } catch (Exception e) {
    
69.             log.error("start tcp server error: {}", e.getMessage());
    
70.         } finally {
    
71.             workGroup.shutdownGracefully();
    
72.             bossGroup.shutdownGracefully();
    
73.         }
    
74.     }
    
75. }

3、Handler



1.  @Slf4j
    
2.  @ChannelHandler.Sharable
    
3.  public class TcpActiveHandler extends ChannelInboundHandlerAdapter {
    

5.      private final static AtomicLong activeCount = new AtomicLong(0);
    

7.      @Override
    
8.      public void channelActive(ChannelHandlerContext ctx) {
    
9.          log.info("active client count: {}", activeCount.addAndGet(1));
    
10.     }
    

12.     @Override
    
13.     public void channelInactive(ChannelHandlerContext ctx) {
    
14.         log.info("active client count: {}", activeCount.decrementAndGet());
    
15.     }
    

17. }



1.  public class TcpMsgOutHandler extends ChannelOutboundHandlerAdapter {
    

3.      @Override
    
4.      public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
    
5.          ctx.writeAndFlush(Unpooled.copiedBuffer(((String) msg).getBytes(StandardCharsets.UTF_8)));
    
6.      }
    
7.  }

netty+4G DTU
zip 0星
超过10%的资源
76KB

下载

 2.  @Slf4j
    
3.  public class TcpMsgInHandler extends ChannelInboundHandlerAdapter {
    

5.      private final ProduceOrderService produceOrderService;
    
6.      private final TestRecordTopService testRecordTopService;
    
7.      private final TestRecordAllService testRecordAllService;
    
8.      private final ProduceOrderLogService produceOrderLogService;
    

10.     public TcpMsgInHandler(ProduceOrderService produceOrderService, TestRecordTopService testRecordTopService,
    
11.  TestRecordAllService testRecordAllService,ProduceOrderLogService produceOrderLogService) {
    
12.         this.produceOrderService = produceOrderService;
    
13.         this.testRecordTopService = testRecordTopService;
    
14.         this.testRecordAllService = testRecordAllService;
    
15.         this.produceOrderLogService = produceOrderLogService;
    
16.     }
    

18.     @Override
    
19.     public void channelActive(ChannelHandlerContext ctx) {
    
20.         log.info("client active: {}", ctx.channel().remoteAddress());
    
21.     }
    

23.     @Override
    
24.     public void channelInactive(ChannelHandlerContext ctx) {
    
25.         log.info("client inactive: {}", ctx.channel().remoteAddress());
    
26.     }
    

28.     @Override
    
29.     public void channelRead(ChannelHandlerContext ctx, Object msg) {
    
30.         ByteBuf byteBuf = (ByteBuf) msg;
    
31.         String readMsg = byteBuf.toString(CharsetUtil.UTF_8);
    
32.         ProduceMsg produceMsg = null;
    
33.         try {
    
34.             produceMsg = JSONObject.parseObject(readMsg, ProduceMsg.class);
    
35.         } catch (Exception e) {
    
36.             log.error("ProduceMsg 格式化消息失败！{}", e.getMessage());
    
37.             ctx.write(ERROR_FORMAT_MSG);
    
38.         }
    

40.         if (produceMsg != null) {
    
41.             log.info("Client Msg: {}", produceMsg);
    
42.             TestRecordTopVO testRecordTopVO;
    
43.             String msgType = produceMsg.getMsgType();
    
44.             String msgContent = produceMsg.getMsgContent();
    
45.             switch (msgType) {
    
46.                 // 通过IMEI号获取订单信息
    
47.                 case FIND_PRODUCE_ORDER_INFO:
    
48.                     ProduceOrderVO produceOrderVO = produceOrderService.queryProduceOrder(msgContent);
    
49.                     if (produceOrderVO == null) {
    
50.                         ctx.write(ERROR_RECORD_NOT_EXIST);
    
51.                     } else {
    
52.                         ctx.write(JSONObject.toJSONString(produceOrderVO));
    
53.                     }
    
54.                     break;
    
55.                 case SAVE_PRODUCE_DATA:
    

57.                     break;
    
58.                 case SAVE_PRODUCE_LOG:
    

60.                     break;
    
61.                 case PRODUCT_OUTBOUND:
    

63.                     break;
    
64.                 case GET_METER_CODE:
    

66.                     break;
    
67.                 case FIND_PRODUCE_RECORD_BY_IMEI:
    

69.                     break;
    
70.                 case FIND_PRODUCE_RECORD_BY_METER_CODE:
    

72.                     break;
    
73.                 default:
    
74.                     log.error("消息不存在！");
    
75.             }
    
76.         }
    

78.     }
    

80.     @Override
    
81.     public void channelReadComplete(ChannelHandlerContext ctx) {
    
82.         ctx.flush();
    
83.     }
    

85.     @Override
    
86.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    
87.         log.error("client {} exception: {}", ctx.channel().remoteAddress(), cause);
    
88.     }
    
89. }