用netty轻松实现一个高效稳定的TCP服务器

          随着物联网的发展，很多项目都开始涉及到了tcp毗连这块，在这里我们轻松用netty去实现,站在巨人的肩膀上。
关于netty包引用：

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

复制代码

实现TCP服务器代码

依赖netty只需几行代码tcp服务：

2. import io.netty.bootstrap.ServerBootstrap;
3. import io.netty.channel.*;
4. import io.netty.channel.nio.NioEventLoopGroup;
5. import io.netty.channel.socket.SocketChannel;
6. import io.netty.channel.socket.nio.NioServerSocketChannel;
7. import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
8. import org.slf4j.Logger;
9. import org.slf4j.LoggerFactory;
11. import java.nio.ByteOrder;
13. public class TcpServer {
15.     private Logger log = LoggerFactory.getLogger(getClass());
16.     //自定义tcp服务端口号
17.     private int port=9000;
19.     static TcpServer tcpServer;
20.     //单例设计模式
21.     private TcpServer(){
23.     }
25.     public static TcpServer getInstance(){
26.         if(tcpServer==null){
27.             tcpServer=new TcpServer();
28.         }
29.         return tcpServer;
30.     };
33. public void run() throws InterruptedException {
36.     // 创建主线程组（接受连接）
37.     EventLoopGroup bossGroup = new NioEventLoopGroup();
38.     // 创建工作线程组（处理连接）
39.     EventLoopGroup workerGroup = new NioEventLoopGroup(20); // 指定工作线程数量为20
41.     // 创建ServerBootstrap实例，用于配置服务器
42.     ServerBootstrap bootstrap = new ServerBootstrap();
43.     // 配置主、工作线程组
44.     bootstrap.group(bossGroup, workerGroup);
45.     // 指定使用NIO进行网络传输
46.     bootstrap.channel(NioServerSocketChannel.class);
47.     // 设置子Channel的Socket选项，允许地址重用
48.     bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);
50.     // 配置子Channel的处理器，这里使用ChannelInitializer
51.     bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
52.         @Override
53.         public void initChannel(SocketChannel ch) throws Exception {
54.             // 添加自定义的解码器,这里是处理协议
55.             ch.pipeline().addLast(new YKCDecoderV1());
56.             // 添加自定义的服务器处理器
57.             ch.pipeline().addLast(new TCPServerHandler());
58.         }
59.     });
61.     // 绑定端口并添加监听器，处理绑定操作的结果
62.     bootstrap.bind(port).addListener((ChannelFutureListener) future -> {
63.         // 在绑定成功后输出日志信息
64.         log.info("bind success in port: " + port);
65.     });
67.     // 输出服务器启动成功信息
68.     System.out.println("server started!");
69. }
73. }

复制代码

 业务处理代码（参考）

以下是处理报文业务类可参考，注意代码未优化：

1. import io.netty.buffer.ByteBuf;
2. import io.netty.buffer.ByteBufUtil;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.handler.codec.ByteToMessageDecoder;
6. import java.net.InetSocketAddress;
7. import java.util.List;
8. import java.util.regex.Matcher;
9. import java.util.regex.Pattern;
11. /**
12. * 正则解析版
13. */
14. public class YKCDecoderV1 extends ByteToMessageDecoder {
16.     final static String reg = "^68.{14,332}";//单指令解析 根据业务协议报文定最短和最长
17.     final static Pattern pattern1 = Pattern.compile(reg);
19.     @Override
20.     protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf bufferIn, List<Object> list) throws Exception {
22.         // 获取可读字节数
23.         int leng = bufferIn.readableBytes();
25.         // 如果可读字节数小于8，输出错误信息并跳过这部分数据
26.         if (leng < 8) {
27.             System.out.println("err! cmd len < 8 .");
28.             String s = ByteBufUtil.hexDump(bufferIn);
29.             System.out.println(s);
30.             bufferIn.skipBytes(leng);
31.             return;
32.         } else {
33.             
34.             String s = ByteBufUtil.hexDump(bufferIn);
35.             Matcher matcher1 = pattern1.matcher(s);
36.             if (matcher1.find()) {
38.                 String cmd = matcher1.group();
39.                 //单指令
40.                 System.out.println("sign cmd: " + cmd);
41.                 String lenStr = cmd.substring(2, 4);
42.                 int len = (Integer.parseInt(lenStr, 16) + 4) * 2;
44.                 int cmdLen = cmd.length();
45.                 if (cmdLen == len) {
46.                     JFYChargeProtocol jfyChargeProtocol = new JFYChargeProtocol(cmd);
47.                     list.add(jfyChargeProtocol);
48.                     bufferIn.skipBytes(leng);
50.                 } else if (cmdLen > len) {
52.                     multiHand(cmd, list);
53.                     bufferIn.skipBytes(leng);
55.                 }
58.             } else {
59.                 logErr(channelHandlerContext, s);
60.                 System.out.println("err! cmd format invalid: " + s);
61.                 bufferIn.skipBytes(leng);
63.             }
65.         }
68.     }
70.     private void multiHand(String cmd, List<Object> list) {
72.         if (cmd.length() < 8) {
73.             return;
74.         }
75.         String lenStr = cmd.substring(2, 4);
76.         int len = (Integer.parseInt(lenStr, 16) + 4) * 2;
78.         if (len > cmd.length()) {
79.             return;
80.         }
82.         String newCmd = cmd.substring(0, len);
84.         if (newCmd.length() == len) {
85.             System.out.println("multi cmd-> " + newCmd);
87.             JFYChargeProtocol jfyChargeProtocol = new JFYChargeProtocol(newCmd);
88.             list.add(jfyChargeProtocol);
89.         }
91.         if (cmd.length() > len) {
92.             System.out.println("multi xxx-> " + cmd);
93.             String two = cmd.substring(len);
94.             if(two.startsWith("68")){
95.                 multiHand(two, list);
96.             }
97.         }
100.     }
102.     private int checkSignCmd(String cmd) {
103.         int cmd_len = getCmdLen(cmd);
104.         return cmd.length() - cmd_len;
105.     }
107.     private int getCmdLen(String cmd) {
108.         String leng = cmd.substring(28, 30) + cmd.substring(26, 28);
109.         int dec_num = Integer.parseInt(leng, 16);
110.         return (dec_num * 2) + 34;
111.     }
113.     private void logErr(ChannelHandlerContext ctx, String msg) {
114.         InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
115.         String clientIP = insocket.getAddress().getHostAddress();
116.         System.out.println(clientIP + " :: " + msg);
117.     }

复制代码

1. public class JFYChargeProtocol {
4.     private int length;
5.     private byte[] raw;
6.     private String rawStr;
8.     public JFYChargeProtocol(int length,byte[] raw){
9.         this.length=length;
10.         this.raw=raw;
11.     }
12.     public JFYChargeProtocol(String raw){
13.         this.rawStr=raw;
14.     }
16.    
18.     public int getLength() {
19.         return length;
20.     }
22.     public void setLength(int length) {
23.         this.length = length;
24.     }
26.     public byte[] getRaw() {
27.         return raw;
28.     }
30.     public void setRaw(byte[] raw) {
31.         this.raw = raw;
32.     }
34.     public String getRawStr() {
35.         return rawStr;
36.     }
38.     public void setRawStr(String rawStr) {
39.         this.rawStr = rawStr;
40.     }
41. }

复制代码

1. import io.netty.buffer.ByteBuf;
2. import io.netty.buffer.Unpooled;
3. import io.netty.channel.ChannelHandlerContext;
4. import io.netty.channel.ChannelInboundHandlerAdapter;
5. import org.apache.logging.log4j.LogManager;
6. import org.apache.logging.log4j.Logger;
7. import org.springframework.stereotype.Service;
8. import java.util.Map;
9. import java.util.concurrent.ConcurrentHashMap;
11. @Service
12. public class TCPServerHandler extends ChannelInboundHandlerAdapter {
13.     private static final Logger logger = LogManager.getLogger(TCPServerHandler.class);
15.     static Map<String,ChannelHandlerContext> inList=new ConcurrentHashMap<String,ChannelHandlerContext>();
18.     /**
19.      * 新连接
20.      * @param ctx
21.      */
22.     @Override
23.     public void channelActive(ChannelHandlerContext ctx) {
24.         String channelName=getChannelName(ctx);
25.         inList.put(channelName,ctx);
26.         logger.info("dev new conn > " +channelName);
27.     }
29.     private String getChannelName(ChannelHandlerContext ctx) {
30.         return "ykc".concat(ctx.channel().remoteAddress().toString());
32.     }
34.     /**
35.      * 连接下线
36.      * @param ctx
37.      * @throws Exception
38.      */
39.     @Override
40.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
41.         String channelName=getChannelName(ctx);
42.         logger.info("dev close conn > " + channelName);
43.         inList.remove(channelName);
44.         ctx.fireChannelInactive();
45.     }
50.     @Override
51.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
52.         JFYChargeProtocol in = (JFYChargeProtocol) msg;
54.         String readMsg= in.getRawStr();
55.         logger.info("read dev <= " + readMsg);
56.         String channelName=getChannelName(ctx);
57.         readMsg=channelName+"$$"+readMsg;
58.         PackageHandlerImpl.getInstance().doHandle(readMsg);
59.         //ctx.writeAndFlush(in);
60.     }
63.     @Override
64.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
65.          cause.printStackTrace();
66.          ctx.close();
67.     }
69.     /**
70.      * 回复信息给设备
71.      * @param hex
72.      */
73.     public static boolean RepDev(String hex){
75.         String[] kv= hex.split("\\$\\$");
76.         if(kv.length==2){
77.             String key=kv[0];
78.             ChannelHandlerContext context=inList.get(key);
79.             if(context!=null){
80.                 byte[] bytes= ByteUtil.hexString2Bytes(kv[1]);
81.                 ByteBuf byteBuf= Unpooled.copiedBuffer(bytes);
82.                 context.writeAndFlush(byteBuf);
83.                 return true;
84.             }else{
85.                 logger.error("dev offline="+key);
86.             }
88.         }else{
89.             logger.error("cmd format err");
91.         }
92.       return false;
93.     }
95. }

复制代码

1. import java.util.ArrayList;
2. import java.util.List;
3. public  class PackageHandlerImpl implements PackageHandler {
5.     public static List<PackageHandler> packageHandlers= new ArrayList<PackageHandler>();
6.     static PackageHandlerImpl packageHandler;
7.     protected PackageHandlerImpl(){
8.         super();
9.         System.out.println("init PackageHandlerImpl");
10.     }
12.     public static PackageHandlerImpl getInstance(){
13.         if(packageHandler==null){
14.             packageHandler=new PackageHandlerImpl();
15.         }
16.         return packageHandler;
17.     }
19.     @Override
20.     public void doHandle(String hex) {
21.         for(PackageHandler f : packageHandlers){
22.             f.doHandle(hex);
23.         }
25.     }
26.     public PackageHandlerImpl addHandle(PackageHandler f){
27.         packageHandlers.add(f);
28.         return this;
29.     }
32. }

复制代码

1. /**
2. * 包处理
3. */
4. public interface PackageHandler {
5.      void doHandle(String hex) ;
7. }

复制代码

1. import org.apache.activemq.command.ActiveMQQueue;
2. import org.apache.logging.log4j.LogManager;
3. import org.apache.logging.log4j.Logger;
4. import org.springframework.beans.factory.annotation.Autowired;
5. import org.springframework.beans.factory.annotation.Value;
6. import org.springframework.jms.annotation.JmsListener;
7. import org.springframework.jms.core.JmsMessagingTemplate;
8. import org.springframework.stereotype.Service;
10. import javax.jms.Destination;
12. @Service
13. public class TranServiceImpl {
15.     private static final Logger logger = LogManager.getLogger(DevServiceApplication.class);
17.     /**
18.      * 接受服务器 返回数据
19.      */
20.     private final String out_name="ykc_out";
23.     /**
24.      * 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
25.      */
26.     @Autowired
27.     private JmsMessagingTemplate jmsTemplate;
29.     /**
30.      * 发送消息 采用系统配置类型
31.      *
32.      * @param queueName 是发送到的队列名称
33.      * @param message   是发送到的队列
34.      */
35.     public void sendMessage(String queueName, final String message) {
36.         jmsTemplate.convertAndSend(queueName, message);
37.     }
39.     /**
40.      * 发送消息 采用指定队列类型
41.      *
42.      * @param queueName 是发送到的队列
43.      * @param message   是发送到的队列
44.      */
45.     public void sendMessageByQueue(String queueName, final String message) {
46.         Destination destination = new ActiveMQQueue(queueName);
47.         jmsTemplate.convertAndSend(destination, message);
48.     }
50.     @JmsListener(destination = out_name)
51.     public void receiveQueue(String text) {
52.         System.out.println("to dev => "+text);
54.         if(!TCPServerHandler.RepDev(text)){
55.             logger.error("write mq fail ==> "+text);
56.         }
57.     }
58. }

复制代码

 运行

当前是集成到 springboot2框架在这里即可运行，或实现实现 org.springframework.boot.ApplicationRunner 或 org.springframework.boot.CommandLineRunner 的接口，即启动后实行的任务，不用框架的在main方法也可以直接运行。

1. /**
2. * tcp服务在框架启动后 跟着启动即可
3. */
4. @SpringBootApplication
5. public class DevServiceApplication {
6.     private static final Logger logger = LogManager.getLogger(DevServiceApplication.class);
9.     public static void main(String[] args) {
12.         TcpServer tcpServer = TcpServer.getInstance();
13.         try {
14.             tcpServer.run();
15.             PackageHandlerImpl packageHandler = PackageHandlerImpl.getInstance();
16.             packageHandler.addHandle(new PackageHandlerByMQ());
18.         } catch (InterruptedException e) {
19.             e.printStackTrace();
20.             logger.error("TCP服务错误", e);
21.             throw new RuntimeException();
22.         }
25.     }

复制代码

总结

看看服务器上的tcp服务运行情况
运行天数：

流量状态图：

            站在netty巨人的肩膀上,这个tcp服务实现方式简单，运行更是稳定。服务器运行时就摆设了不绝到今天共运行1235天了，900多个装备同步在线，配了2g的jvm运行内存，cpu占用5.6（top截图等了很久才出来5.6是个峰值，平常不到1）确保某个市的充电桩装备。中心由于客户的充电桩装备协议题目更新过几次，刚时开始是使用netty官网的解码LengthFieldBasedFrameDecoder做处理，可以说非常高效，但随后发现有几个产商的装备报文头部有特别字符，而且刚好和协议头有些重叠，再考虑到示来的产商协议的不确定性，为了兼容这些产家不得以并以正则的方法行止理。

扩展部分

Netty 官方提供的编解码器

• 字符串编解码器：
  
  
  
  • StringEncoder：将字符串编码为字节。
    
  • StringDecoder：将字节解码为字符串。
    
  
  
• 字节流编解码器：
  
  
  
  • ByteArrayEncoder：将字节数组编码为字节。
    
  • ByteArrayDecoder：将字节解码为字节数组。
    
  
  
• 对象序列化编解码器：
  
  
  
  • ObjectEncoder：将对象序列化为字节。
    
  • ObjectDecoder：将字节反序列化为对象。
    
  
  
• 长度字段编解码器：
  
  
  
  • LengthFieldPrepender：在消息头部添加表现消息长度的字段。
    
  • LengthFieldBasedFrameDecoder：根据长度字段解码消息，用于处理拆包和粘包题目。
    
  
  
• 行分隔符编解码器：
  
  
  
  • LineBasedFrameDecoder：按行切分消息，通常用于处理文本协议。
    
  
  
• DelimiterBasedFrameDecoder：
  
  
  
  • DelimiterBasedFrameDecoder：按照指定的分隔符切分消息，用于处理自定义分隔符的协议。
    
  
  
• Protobuf 编解码器：
  
  
  
  • ProtobufEncoder：将 Protobuf 对象编码为字节。
    
  • ProtobufDecoder：将字节解码为 Protobuf 对象。
    
  
  
• HTTP 编解码器：
  
  
  
  • HttpRequestEncoder：将 HTTP 哀求编码为字节。
    
  • HttpResponseDecoder：将字节解码为 HTTP 响应。
    
  • HttpRequestDecoder：将字节解码为 HTTP 哀求。
    
  • HttpResponseEncoder：将 HTTP 响应编码为字节。
    
  
  
• WebSocket 编解码器：
  
  
  
  • WebSocketServerProtocolHandler：处理 WebSocket 握手以及帧的编解码。
    
  
  

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