186 lines
6.3 KiB
Markdown
186 lines
6.3 KiB
Markdown
# TCP通信流程详解
|
||
|
||
## 核心问题解答
|
||
|
||
**问题：Message对象是设备直接回复的，还是底层框架封装的？**
|
||
|
||
**答案：Message对象是底层框架封装的，不是设备直接回复的。**
|
||
|
||
设备只发送原始的字节数据（ByteBuf），框架负责：
|
||
1. 接收设备的原始字节数据
|
||
2. 通过协议解码器解析为Message对象
|
||
3. 匹配请求和响应
|
||
4. 将响应传递给等待的request
|
||
|
||
## 完整通信流程
|
||
|
||
### 1. 平台下发指令（requestR调用）
|
||
|
||
```
|
||
PowerOnTriggerTaskExecutor.execute()
|
||
↓
|
||
MessageManager.requestR(serialNumber, msg, Message.class)
|
||
↓
|
||
Session.request(request, Message.class)
|
||
├─ 生成requestKey（基于Message类型和serNo）
|
||
├─ 订阅等待响应（subscribe(key)）
|
||
└─ 发送指令到设备（channel.writeAndFlush()）
|
||
```
|
||
|
||
**关键代码：**
|
||
```java
|
||
// 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()
|
||
↓ 消息分发处理
|
||
```
|
||
|
||
**关键代码：**
|
||
```java
|
||
// 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）负责：
|
||
1. 解析原始字节数据
|
||
2. 识别是否是设备回复（通过协议特征，如Modbus功能码）
|
||
3. 设置isReply标志
|
||
4. 封装为DeviceReport对象
|
||
|
||
**关键代码：**
|
||
```java
|
||
// ModbusProtocol.decode()
|
||
case Write06:
|
||
case Write05:
|
||
// 如果返回06编码，说明是设备回复
|
||
report.setIsReply(true); // 标记为回复消息
|
||
report.setStatus(FunctionReplyStatus.SUCCESS);
|
||
return report;
|
||
```
|
||
|
||
### 5. 响应匹配机制（关键！）
|
||
|
||
**问题：设备回复如何匹配到等待的request？**
|
||
|
||
**答案：通过Session的requestKey和responseKey机制**
|
||
|
||
```java
|
||
// 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;
|
||
}
|
||
```
|
||
|
||
**匹配流程：**
|
||
1. 发送请求时，生成requestKey并订阅
|
||
2. 设备回复后，协议解码为Message对象
|
||
3. **需要调用Session.response(message)**来匹配
|
||
4. 如果key匹配，Mono会收到响应，requestR返回成功
|
||
|
||
### 6. 当前代码的问题（重要发现！）
|
||
|
||
**重要发现：在当前代码中，Session.response()方法没有被调用！**
|
||
|
||
这意味着：
|
||
- 设备回复的Message对象被解码后，**没有被匹配到等待的request**
|
||
- requestR()的Mono会一直等待，直到超时
|
||
- **这就是为什么你的代码中，即使设备回复了，requestR()也可能返回超时**
|
||
|
||
**实际流程：**
|
||
1. requestR()发送请求，订阅等待响应
|
||
2. 设备回复原始字节数据
|
||
3. 协议解码器解码为DeviceReport对象，设置isReply=true
|
||
4. **但是，DispatcherHandler将消息发送到MQ，而不是调用Session.response()**
|
||
5. requestR()的Mono一直等待，200ms后超时
|
||
|
||
**关键代码：**
|
||
```java
|
||
// 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()是否能正确返回？
|
||
|
||
## 总结
|
||
|
||
1. **设备只发送原始字节数据**，不知道Message对象
|
||
2. **框架负责解码**：ByteBuf → Message对象
|
||
3. **协议解码器识别回复**：通过协议特征（如Modbus功能码）设置isReply标志
|
||
4. **响应匹配机制**：通过requestKey/responseKey匹配请求和响应
|
||
5. **关键问题**：需要确认Session.response()是否被正确调用
|
||
|
||
## 建议
|
||
|
||
对于上电触发任务，如果Session.response()没有被调用，那么：
|
||
- requestR()可能无法正确识别设备回复
|
||
- 需要检查是否有其他地方调用了Session.response()
|
||
- 或者需要修改代码，在协议解码后主动调用Session.response()
|
||
|