使用netty作为TCP服务端并使用16进制与硬件进行通信

字数统计: 3.2k字   |   阅读时长≈ 14分

说明

本文记录了在使用netty作为TCP服务端的情况下，如何解决TCP粘包、半包问题，以及如何使用16进制与客户端进行通信。

依赖

首先添加maven依赖：

1
2
3
4
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
</dependency>

代码

创建字节转换类

首先创建一个将字节转换为16进制的处理类，并判断包头包尾是否正确。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
public class CustomDecode extends ByteToMessageDecoder {
	
	Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
    	int len = in.readableBytes();       //这里得到可读取的字节长度
        in.markReaderIndex();               //包头做标记位，后面可以重新回到数据包头开始读数据
        //有数据时开始读数据包
        if (len > 0) {
            byte[] src = new byte[len];
            in.readBytes(src);          //把数据读到字节数组中(读取完之后指针会到最后一个数据)
            in.resetReaderIndex();      //重置当前指针到标记位(包头),用于重新读取接收的数据，直至接收完完整数据包
            if(len>1) {
            	//判断包头是否为约定的字节
            	if((src[0] & 0x000000ff) != 0xAB || (src[1] & 0x000000ff) != 0xCD) {
            		logger.warn("无法识别的包头，转换为16进制的包头为：{}",StringUtil.byteToHexString(src));
            		// 包头不对，直接断开连接
                	channelHandlerContext.close();
            	}
            	//字节转换为16进制，判断最后两个字节是否为结尾字节,0x000000ff等于十进制的255
            	if((src[len-2] & 0x000000ff) == 0xEF && (src[len-1] & 0x000000ff) == 0xGG) {
            		in.skipBytes(in.readableBytes());//标记已读取完毕
            		//收到了尾部字节，完成接收
            		out.add(src);
                }
            }
          //非结尾字节，继续接收,当数据包的长度不够时直接return，netty在缓冲区有数据时会一直调用decode方法，所以我们只需要等待下一个数据包传输过来一起解析(解决半包问题)
            return;
        }else {
        	// 发空数据包过来的家伙直接断开连接
        	channelHandlerContext.close();
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    	logger.error("TCP连接异常！，信息：{}",cause.getMessage(),cause);
//        ctx.close();
    }
}

以上为数据处理类，首先判断包头是否为约定的16进制数据，其次判断包尾是否为约定的字节，如不是则return继续接收数据直到接收到了包尾。

工具类 StringUtil.byteToHexString的作用是将字节转换为16进制字符串，此工具类为netty自带，**(src[0] & 0x000000ff) 表示将下标为0的字节转换为16进制**

创建字节处理初始化类

1
2
3
4
5
6
7
8
9
10
11
12
13
public class ServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        channel.pipeline(). 
		        //包尾以EFGG结束，使用netty自带的粘包处理器,false参数表示不去掉包尾字符
        		addLast(new DelimiterBasedFrameDecoder(1024,false,Unpooled.copiedBuffer(TCPServerUtils.hexStr2bytes("EFGG")))).
                addLast(new CustomDecode()).        //自定义解码器
                addLast(new TCPServerHandler())     //自定义处理器
        ;
    }
    
}

初始化字节处理类，并使用netty自带的粘包处理器解决粘包问题，addLast(new DelimiterBasedFrameDecoder(1024,false,Unpooled.copiedBuffer(TCPServerUtils.hexStr2bytes("EFGG")))) 表示以EFGG结尾的数据分割包

简单的说明一下半包和粘包，半包： 假设数据包以16进制 AABB开头，以EFGG结尾，中间放具体数据，但是TCP传输可能只传了AABB，EFGG可能是在第二次传输过来，因此要等一个完整的数据包传输过来后再做处理。

粘包： 假设数据包内容为”AABBHHEFGG“是一个完整的包，但是TCP在传输过程中可能是这样传的：AABBHHEFGGAABBH ,本来应该是两个包的内容给放在了一起，而且还不完整，这就是粘包。

创建业务处理类

有了上面的字节处理类，在这一层拿到的就是分割好的完整数据包了，此时可以对字节做截取，去掉包头包尾读取中间内容，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
public class TCPServerHandler extends ChannelInboundHandlerAdapter{
	Logger logger = LoggerFactory.getLogger(getClass());

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		// 这里msg就是从解码器中传来的数据，解码器传输过来是什么格式，这里直接转成对应的格式就可以
		byte[] src = (byte[]) msg;

		String hexString = StringUtil.toHexString(src);
		//logger.info("接收到完整16进制数据：{}", hexString);
		try {
			Channel channel = ctx.channel();

			// TODO 此处一般先要拿到数据包做一次checksum，即判断一下包数据是否正确
			
	
			
			// 要返回的字节
			byte[] resp = null;

			
			// 假设功能码字节在第8位
			byte action = src[8];
			
			switch ((action & 0x000000ff)) {
			case 0x08:
				// 假设0x08表示请求认证
				resp = ServerUtils.makeResponse("0200","01");// 返回成功
				
				//TODO 此处绑定channel与识别码
				map.put(deviceId,channel);//deviceId 从数据包中截取

				//上报在线状态
				...
				break;

			
			default:
				logger.error("未知的上传包信息！");
				break;
			}
			
			if (resp != null) {
				ByteBuf buf = Unpooled.copiedBuffer(resp);
				channel.writeAndFlush(buf);
			}

		} catch (Exception e) {
			logger.info("服务器处理出错！出错信息：{}",e.getMessage(),e);
		}
	}
	
	
	/**
	 * 设备离线事件
	 */
	@Override
	public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
		//TODO 上报离线状态
		super.handlerRemoved(ctx);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		cause.printStackTrace();
		logger.error("发生异常:" + cause.getMessage());

	}

	
}

注意点：一定要新建一个Map保存认证设备的唯一识别码，netty中的channel即表示一个会话，因此在请求认证时就要将识别码与channel进行绑定，这样才可以随时向客户端发送消息。

checksum工具类代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public static String makeChecksum(String data) {
	if (data == null || data.equals("")) {
		return "";
	}
	int total = 0;
	int len = data.length();
	int num = 0;
	while (num < len) {
		String s = data.substring(num, num + 2);
		total += Integer.parseInt(s, 16);
		num = num + 2;
	}
	/**
	 * 用256求余最大是255，即16进制的FF
	 */
	int mod = total % 256;
	String hex = Integer.toHexString(mod);
	len = hex.length();
	// 如果不够校验位的长度，补0,这里用的是两位校验
	if (len < 2) {
		hex = "0" + hex;
	}
	return hex;
}

创建netty TCP服务启动类

由于我是使用的spring boot框架因此直接在初始化事件中创建netty服务，也可以直接在main方法中启动，代码如下 ：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public class ServerStarter implements ApplicationListener<ContextRefreshedEvent>{
	
	Logger logger = LoggerFactory.getLogger(getClass());

	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		 //创建线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建启动类
            ServerBootstrap b = new ServerBootstrap(); 
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ServerInitializer())
                    .option(ChannelOption.SO_BACKLOG, 256)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            // 绑定端口
            ChannelFuture f = b.bind(2077).sync();
            // 等待服务器 socket 关闭 。
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
		
	}
	
}

OK,现在TCP服务的架子已经搭好，只需往里面填充业务代码即可。

常用的工具类

附上可能会用到的工具类，比如16进制转10进制，16进制转2进制等

十六进制转byte[]数组

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public static byte[] hexStr2bytes(String hexStr) {
	if (StringUtils.isEmpty(hexStr)) {
		return null;
	}
	if (hexStr.length() % 2 != 0) {// 长度为单数
		hexStr = "0" + hexStr;// 前面补0
	}
	char[] chars = hexStr.toCharArray();
	int len = chars.length / 2;
	byte[] bytes = new byte[len];
	for (int i = 0; i < len; i++) {
		int x = i * 2;
		bytes[i] = (byte) Integer.parseInt(String.valueOf(new char[] { chars[x], chars[x + 1] }), 16);
	}
	return bytes;
}

大小端处理 ，低位字节排放在内存的低地址端，高位字节排放在内存的高地址端。

1
2
3
public static int toLittleEndian(int a) {
    return (((a & 0xFF) << 24) | (((a >> 8) & 0xFF) << 16) | (((a >> 16) & 0xFF) << 8) | ((a >> 24) & 0xFF));
}

10进制数字转16进制数字

1
2
3
4
5
6
7
public static String numberToHexString(Integer number) {
		String hex = Integer.toHexString(number);
		if(number<16) {
			hex = "0"+hex;
		}
		return hex;
	}

checksum，上面已经贴过再贴一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public static String makeChecksum(String data) {
	if (data == null || data.equals("")) {
		return "";
	}
	int total = 0;
	int len = data.length();
	int num = 0;
	while (num < len) {
		String s = data.substring(num, num + 2);
		total += Integer.parseInt(s, 16);
		num = num + 2;
	}
	/**
	 * 用256求余最大是255，即16进制的FF
	 */
	int mod = total % 256;
	String hex = Integer.toHexString(mod);
	len = hex.length();
	// 如果不够校验位的长度，补0,这里用的是两位校验
	if (len < 2) {
		hex = "0" + hex;
	}
	return hex;
}

截取字节 （这个可能是最常用的了，后来我才发现java自带的bytebuffer已经自带大小端转换功能）

1
2
3
4
5
6
7
8
9
10
11
12
public static String catoutByte(byte[] sourceBytes,int startPos,int len) {
	byte[] cbyte = new byte[len]; 
	System.arraycopy(sourceBytes, startPos, cbyte, 0, len);//截取字节
	//使用Java bytebuff处理类包装
	ByteBuffer bb = ByteBuffer.wrap(cbyte);
	
	//大小端转换,转换成小端模式
	bb.order(ByteOrder.LITTLE_ENDIAN);
	String vHex = StringUtil.toHexString(bb.array());
	return vHex;

}

使用Redis作为设备指令发送队列

更新日期：2021-7-7 ，新增使用redis作简单队列，发送指令到设备。

此方式需要可容忍指令丢失，因为有可以在取出指令后服务器挂掉或一直未获取到锁导致指令超时，可靠消息列队应当使用队列中间件

首先在你的发送指令类的构造函数中创建一个线程池，在指令并发较大时可以有效节省资源

1
2
3
4
5
6
7
ExecutorService pool;

public Commander() {
	pool = Executors.newFixedThreadPool(35);//参数看情况具体设定
	//检查队列命令专用线程,该线程不可结束
	pool.execute(()->sendCommandByQueen(false));
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

	/**
	 * @Title: sendCommandByQueue
	 * @Description:   从队列中获取指令并排队发送
	 * @author reiner_shir
	 * @date 2021年7月7日
	 * @param canOver 是否可以在没有指令时结束循环
	 * @return
	 */
	private void sendCommandByQueue(boolean canOver) {
		do {
				//从队列中取指令
				String commandStr = canOver?redisTemplate.opsForList().rightPop(REDIS_COMMAND_LIST):
						redisTemplate.opsForList().rightPop(REDIS_COMMAND_LIST,0,TimeUnit.SECONDS);
				if(StringUtils.isEmpty(commandStr)) {
					if(canOver) {
						break;
					}
				}
				//读取队列中的命令
				Command command=null;
				try {
                    //json转对象
					command = JacksonUtil.readValue(commandStr, Command.class);
				} catch (Exception e) {
					logger.error("指令内容转换出错：{}",command,e);
					continue;
				}
				String deviceId = command.getDeviceId();
				String actionCode = command.getActionCode();
				String responseData= command.getResponseData();
				long time = command.getTime();
				if(!NettyChannelManager.hasKey(deviceId)) {
					logger.warn("错误的指令请求，该设备未连接，设备号：{}",deviceId);
					continue;
				}
				String key = Contract.COMMAND_REDIS_KEY+deviceId;
				pool.execute(()->{
					boolean flag = false;
					do {
						long nowTime = System.currentTimeMillis();
						//判断队列中的指令是否超时
						if(((nowTime-time)/1000)>15) {
							logger.warn("指令已超时：指令内容：{},功能码：{},发送的设备号：{}，超时时间：{}",responseData,actionCode,deviceId,((nowTime-time)/1000));
							break;
						}
						//判断该设备是否有正在发送中的指令,指令状态失效时间为7秒
						if(redisTemplate.opsForValue().setIfAbsent(key, "1", 7, TimeUnit.SECONDS)) {
							//deviceId = command.getDeviceId();
							
							//如果没有正在发送中的指令，占用锁发送指令
							sendCommand(deviceId,actionCode,responseData);
							flag = true;
						}
						try {
							//否则睡眠一秒再检查
							Thread.sleep(400l);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}while(!flag);
				});
		}while(true);
	}
	
	private final String REDIS_COMMAND_LIST="REDIS_COMMAND_QUEUE_LIST";
	
/**
*
*对外调用此方法
*/
	public boolean pushCommand(String deviceId,String responseData) {
		Command command = new Command();
		command.setDeviceId(deviceId);
		command.setResponseData(responseData);
		command.setTime(new Date().getTime());
		//放入队列
		boolean r = redisTemplate.opsForList().leftPush(REDIS_COMMAND_LIST, JacksonUtil.toJSon(command))>0?true:false;
		//每超过10条指令创建一个可结束的执行指令线程
		long size = redisTemplate.opsForList().size(REDIS_COMMAND_LIST);
		if(size>=20&&size%20==0) {
			logger.info("当前队列指令条数：{} mod:{}",size,size%20);
			pool.execute(()->sendCommandByQueue(true));
		}
		return r;
	}
	
	private boolean sendCommand(String deviceId,String responseData) {
		Channel channel = NettyChannelManager.getChannel(deviceId);
		if(channel!=null) {
            //将16进制指令转换成字节
			byte[] resp = TCPServerUtils.makeResponse(responseData);
			
			//把发送的消息缓存起来，等待客户端的响应
			NettyChannelManager.putSendMessage(channel,resp,responseData);
			
			//发送指令
			channel.writeAndFlush(resp);
			
			//每1.5秒后检查是否已经收到了回复,共检查3次
			Timer timer = new Timer();
			timer.schedule(new TimerTask() {
				@Override
				public void run() {
					SendMessage checkMessage = NettyChannelManager.getSendMessageByParamter(channel);
					if(checkMessage!=null) {
						if (checkMessage.getTimes() <= 3) 
						{
							checkMessage.setTimes(checkMessage.getTimes()+1);
							byte[] data = (byte[]) checkMessage.getData();
							//如果没有收到回复，则重新发送
							TCPServerUtils.writeAndFlush(channel, data);
						}else {
                            //移除发送中的消息
							NettyChannelManager.removeSendMessageAndRedisMsg(channel);
							//超过3次就不管了
							timer.cancel();
						}
					}else {
						timer.cancel();
					}
				}
			}, 3000l,1500l);//3秒后才检查是为了留出上报状态的时间，上报成功后会移除SendMessage
			
			
			return true;
		}
		
		return false;
	}




可调用public boolean pushCommand(String deviceId,String responseData) 方法将指令放入队列，等待获取指令列队的线程执行。

指令实体类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
public class Command {

	private String deviceId;
	private String actionCode;
	private String responseData;
	private Long time;
	public String getDeviceId() {
		return deviceId;
	}
	public void setDeviceId(String deviceId) {
		this.deviceId = deviceId;
	}
	
	public String getActionCode() {
		return actionCode;
	}
	public void setActionCode(String actionCode) {
		this.actionCode = actionCode;
	}
	public String getResponseData() {
		return responseData;
	}
	public void setResponseData(String responseData) {
		this.responseData = responseData;
	}
	public Long getTime() {
		return time;
	}
	public void setTime(Long time) {
		this.time = time;
	}
	
	
	
}

结尾

最后一个注意点，一定要做重试机制，TCP传输过程中很可能会丢包，我目前的做法是在发送消息成功后保存一个临时的缓存，当接收到客户端回复后将缓存移除，如若一直未接收到回复（比如1秒后）则再次重新发送。

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

关于我：多年工作经验，熟悉微高并发服务架构设计与开发，有AI客户机器人开发相关经验。
项目或内推机会请联系邮箱： [email protected]
Update By 2023-6-14