6.3 KiB
6.3 KiB
TCP通信流程详解
核心问题解答
问题:Message对象是设备直接回复的,还是底层框架封装的?
答案:Message对象是底层框架封装的,不是设备直接回复的。
设备只发送原始的字节数据(ByteBuf),框架负责:
- 接收设备的原始字节数据
- 通过协议解码器解析为Message对象
- 匹配请求和响应
- 将响应传递给等待的request
完整通信流程
1. 平台下发指令(requestR调用)
PowerOnTriggerTaskExecutor.execute()
↓
MessageManager.requestR(serialNumber, msg, Message.class)
↓
Session.request(request, Message.class)
├─ 生成requestKey(基于Message类型和serNo)
├─ 订阅等待响应(subscribe(key))
└─ 发送指令到设备(channel.writeAndFlush())
关键代码:
// Session.request()方法
public <T> Mono<T> request(Message message, Class<T> responseClass) {
String key = requestKey(message, responseClass); // 生成匹配key
Mono<T> subscribe = this.subscribe(key); // 订阅等待响应
ChannelFuture channelFuture = channel.writeAndFlush(Packet.of(this, message));
return Mono.create(...).then(subscribe); // 等待设备回复
}
2. 设备发送原始数据
设备通过TCP连接发送原始字节数据(ByteBuf),例如:
- Modbus RTU协议:
7E 01 03 00 00 00 02 C4 0B 7E - 设备不知道Message对象,只发送协议规定的字节流
3. 框架接收和处理(Netty Handler链)
TCPMessageAdapter.channelRead()
↓ 接收原始ByteBuf
MessageDecoderWrapper.channelRead()
↓ 协议解码:ByteBuf → DeviceReport(Message)
DispatcherHandler.channelRead()
↓ 消息分发处理
关键代码:
// TCPMessageAdapter - 步骤1:接收原始数据
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf buf = (ByteBuf) msg; // 设备发送的原始字节
Session session = getSession(ctx);
ctx.fireChannelRead(Packet.of(session, buf)); // 传递给下一个处理器
}
// MessageDecoderWrapper - 步骤2:协议解码
public void channelRead(ChannelHandlerContext ctx, Object msg) {
Packet packet = (Packet) msg;
ByteBuf input = packet.take();
DeviceReport message = decoder.decode(input, packet.session.getClientId());
// 协议解码器将ByteBuf解析为DeviceReport对象
ctx.fireChannelRead(packet.replace(message));
}
// DispatcherHandler - 步骤3:消息分发
public void channelRead(ChannelHandlerContext ctx, Object msg) {
Packet packet = (Packet) msg;
Message request = packet.message; // 已经是Message对象了
// 处理消息...
}
4. 协议解码器的工作
协议解码器(如ModbusProtocol)负责:
- 解析原始字节数据
- 识别是否是设备回复(通过协议特征,如Modbus功能码)
- 设置isReply标志
- 封装为DeviceReport对象
关键代码:
// ModbusProtocol.decode()
case Write06:
case Write05:
// 如果返回06编码,说明是设备回复
report.setIsReply(true); // 标记为回复消息
report.setStatus(FunctionReplyStatus.SUCCESS);
return report;
5. 响应匹配机制(关键!)
问题:设备回复如何匹配到等待的request?
答案:通过Session的requestKey和responseKey机制
// Session.request() - 发送请求时
private static String requestKey(Message request, Class responseClass) {
String className = responseClass.getName();
if (Response.class.isAssignableFrom(responseClass)) {
String serNo = request.getSerNo(); // 使用序列号
return className + '.' + serNo;
}
return className; // 普通Message使用类名
}
// Session.response() - 接收响应时
private static String responseKey(Object response) {
String className = response.getClass().getName();
if (response instanceof Response) {
int serialNo = ((Response) response).getResponseSerialNo();
return className + '.' + serialNo;
}
return className;
}
匹配流程:
- 发送请求时,生成requestKey并订阅
- 设备回复后,协议解码为Message对象
- **需要调用Session.response(message)**来匹配
- 如果key匹配,Mono会收到响应,requestR返回成功
6. 当前代码的问题(重要发现!)
重要发现:在当前代码中,Session.response()方法没有被调用!
这意味着:
- 设备回复的Message对象被解码后,没有被匹配到等待的request
- requestR()的Mono会一直等待,直到超时
- 这就是为什么你的代码中,即使设备回复了,requestR()也可能返回超时
实际流程:
- requestR()发送请求,订阅等待响应
- 设备回复原始字节数据
- 协议解码器解码为DeviceReport对象,设置isReply=true
- 但是,DispatcherHandler将消息发送到MQ,而不是调用Session.response()
- requestR()的Mono一直等待,200ms后超时
关键代码:
// DispatcherHandler.handleReport() - 设备回复被发送到MQ
private void handleReport(Message request) {
DeviceReport report = (DeviceReport) request;
DeviceReportBo reportBo = new DeviceReportBo();
// ...
reportBo.setIsReply(report.getIsReply()); // 标记为回复
MessageProducer.sendPublishMsg(reportBo); // 发送到MQ,而不是Session.response()
}
这意味着:
- requestR()机制在当前实现中可能无法正确工作
- 设备回复通过MQ异步处理,而不是同步匹配到request
- 你的代码中检查AjaxResult的code和data是正确的方向
- 但需要确认:如果设备回复了,requestR()是否能正确返回?
总结
- 设备只发送原始字节数据,不知道Message对象
- 框架负责解码:ByteBuf → Message对象
- 协议解码器识别回复:通过协议特征(如Modbus功能码)设置isReply标志
- 响应匹配机制:通过requestKey/responseKey匹配请求和响应
- 关键问题:需要确认Session.response()是否被正确调用
建议
对于上电触发任务,如果Session.response()没有被调用,那么:
- requestR()可能无法正确识别设备回复
- 需要检查是否有其他地方调用了Session.response()
- 或者需要修改代码,在协议解码后主动调用Session.response()