diff --git a/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/ConnectionTraceHandler.java b/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/ConnectionTraceHandler.java new file mode 100644 index 00000000..0ec96e82 --- /dev/null +++ b/fastbee-server/base-server/src/main/java/com/fastbee/base/core/hanler/ConnectionTraceHandler.java @@ -0,0 +1,82 @@ +package com.fastbee.base.core.hanler; + +import com.fastbee.common.trace.TraceContext; +import com.fastbee.common.trace.TraceIdGenerator; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInboundHandlerAdapter; +import io.netty.util.AttributeKey; + +/** + * 连接级链路追踪处理器(TCP 专用) + * + *

与 {@link TraceHandler}(每报文一个 TraceId)不同,本处理器以「一次 TCP 连接」为粒度: + * 在连接建立(channelActive)时生成唯一 TraceId 并绑定到 Channel 的 attr 上, + * 该连接生命周期内的所有入站事件(读、异常、空闲等)共用同一个 TraceId, + * 便于把同一设备 TCP 会话的全部日志用一个 traceId 检索出来。 + * + *

注意: + *

+ * + * @author FastBee + */ +@ChannelHandler.Sharable +public class ConnectionTraceHandler extends ChannelInboundHandlerAdapter { + + /** + * 共享实例(无状态,可跨 Channel 复用) + */ + public static final ConnectionTraceHandler INSTANCE = new ConnectionTraceHandler(); + + /** + * 连接级 TraceId 在 Channel 属性中的 key + */ + private static final AttributeKey CONN_TRACE_ID = AttributeKey.valueOf("connTraceId"); + + @Override + public void channelActive(ChannelHandlerContext ctx) { + // 连接建立时生成一次连接级 TraceId,绑定到当前 Channel + ctx.channel().attr(CONN_TRACE_ID).set(TraceIdGenerator.generate()); + withTrace(ctx, ctx::fireChannelActive); + } + + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) { + withTrace(ctx, () -> ctx.fireChannelRead(msg)); + } + + @Override + public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { + withTrace(ctx, () -> ctx.fireUserEventTriggered(evt)); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + withTrace(ctx, () -> ctx.fireExceptionCaught(cause)); + } + + /** + * 包裹一次入站事件:套用连接级 TraceId 到 MDC,执行后清理, + * 防止 EventLoop 线程复用导致 TraceId 串到其它连接。 + * + * @param ctx 当前 Channel 上下文 + * @param event 需要包裹的入站事件 + */ + private void withTrace(ChannelHandlerContext ctx, Runnable event) { + String traceId = ctx.channel().attr(CONN_TRACE_ID).get(); + if (traceId != null) { + TraceContext.setTraceId(traceId); + } + try { + event.run(); + } finally { + TraceContext.clear(); + } + } +} diff --git a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java index d5ddb77a..709c8a00 100644 --- a/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java +++ b/fastbee-server/iot-server-core/src/main/java/com/fastbee/server/TCPServer.java @@ -1,6 +1,6 @@ package com.fastbee.server; -import com.fastbee.base.core.hanler.TraceHandler; +import com.fastbee.base.core.hanler.ConnectionTraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.server.config.NettyConfig; import com.fastbee.server.handler.*; @@ -63,7 +63,8 @@ public class TCPServer extends Server { protected void initChannel(NioSocketChannel channel) throws Exception { channel.pipeline() // 第一步、链路追踪(必须位于编解码器之前,让解码日志也带 TraceId) - .addLast(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) + // 连接级:同一 TCP 连接生命周期内共用一个 TraceId,便于整段会话检索 + .addLast(FastBeeConstant.SERVER.TRACE, ConnectionTraceHandler.INSTANCE) // 设置心跳时间 .addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime)) // 第二步、添加粘包处理器 diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java index 6708dc56..26573753 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/adapter/MqttMessageDelegate.java @@ -56,7 +56,7 @@ public class MqttMessageDelegate { throw new ServiceException("客户端未连接"); } - /*TraceId 由 pipeline 头部的 TraceHandler 统一生成与清理*/ + /*TraceId 由 pipeline 头部的 ConnectionTraceHandler(连接级)统一生成与清理*/ Optional.of(processor.get(messageType)) .ifPresent(mqttHandler -> mqttHandler.handler(ctx, message)); } diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java index 7fe40f53..afe2c88f 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/MqttServer.java @@ -1,6 +1,6 @@ package com.fastbee.mqtt.server; -import com.fastbee.base.core.hanler.TraceHandler; +import com.fastbee.base.core.hanler.ConnectionTraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; import com.fastbee.server.Server; @@ -56,7 +56,7 @@ public class MqttServer extends Server { .addFirst(FastBeeConstant.SERVER.IDLE , new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime, TimeUnit.SECONDS)) // 链路追踪(位于编解码器之前,让解码日志也带 TraceId) - .addFirst(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) + .addFirst(FastBeeConstant.SERVER.TRACE, ConnectionTraceHandler.INSTANCE) .addLast(FastBeeConstant.SERVER.DECODER, new MqttDecoder(1024 * 1024 * 2)) .addLast(FastBeeConstant.SERVER.ENCODER, MqttEncoder.INSTANCE) .addLast(messageAdapter); diff --git a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java index a7962116..d37dffb4 100644 --- a/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java +++ b/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/server/WebSocketServer.java @@ -1,6 +1,6 @@ package com.fastbee.mqtt.server; -import com.fastbee.base.core.hanler.TraceHandler; +import com.fastbee.base.core.hanler.ConnectionTraceHandler; import com.fastbee.common.constant.FastBeeConstant; import com.fastbee.mqtt.codec.WebSocketMqttCodec; import com.fastbee.mqtt.handler.adapter.MqttMessageAdapter; @@ -51,7 +51,7 @@ public class WebSocketServer extends Server { .addFirst(FastBeeConstant.WS.HEART_BEAT , new IdleStateHandler(0, 0, 70)) /*链路追踪(位于编解码器之前,让解码日志也带 TraceId)*/ - .addFirst(FastBeeConstant.SERVER.TRACE, TraceHandler.INSTANCE) + .addFirst(FastBeeConstant.SERVER.TRACE, ConnectionTraceHandler.INSTANCE) /*http请求响应*/ .addLast(FastBeeConstant.WS.HTTP_SERVER_CODEC, new HttpServerCodec()) /*聚合header与body组成完整的Http请求,最大数据量为1Mb*/