TraceId TCP上报调整为一次连接一个traceId
This commit is contained in:
zhugongping
parent
edb65ca44e
commit
d84491fe77
@ -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 专用）
|
||||
*
|
||||
* <p>与 {@link TraceHandler}（每报文一个 TraceId）不同，本处理器以「一次 TCP 连接」为粒度：
|
||||
* 在连接建立（channelActive）时生成唯一 TraceId 并绑定到 Channel 的 attr 上，
|
||||
* 该连接生命周期内的所有入站事件（读、异常、空闲等）共用同一个 TraceId，
|
||||
* 便于把同一设备 TCP 会话的全部日志用一个 traceId 检索出来。
|
||||
*
|
||||
* <p>注意：
|
||||
* <ul>
|
||||
* <li>Netty EventLoop 线程被多连接复用，因此每次入站事件处理前后必须 set/clear MDC，
|
||||
* 避免 TraceId 在 EventLoop 上串线。连接级仅指「TraceId 的取值来源」，
|
||||
* 不再每次重新生成，清理逻辑与 {@link TraceHandler} 一致。</li>
|
||||
* <li>TraceId 存储在 Channel.attr 中，handler 自身保持无状态，可继续用 {@code @Sharable} 单例。</li>
|
||||
* <li>UDP 为无连接协议，请勿复用本处理器，仍应使用 {@link TraceHandler}（每报文一个 TraceId）。</li>
|
||||
* </ul>
|
||||
*
|
||||
* @author FastBee
|
||||
*/
|
||||
@ChannelHandler.Sharable
|
||||
public class ConnectionTraceHandler extends ChannelInboundHandlerAdapter {
|
||||
|
||||
/**
|
||||
* 共享实例（无状态，可跨 Channel 复用）
|
||||
*/
|
||||
public static final ConnectionTraceHandler INSTANCE = new ConnectionTraceHandler();
|
||||
|
||||
/**
|
||||
* 连接级 TraceId 在 Channel 属性中的 key
|
||||
*/
|
||||
private static final AttributeKey<String> 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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))
|
||||
// 第二步、添加粘包处理器
|
||||
|
||||
@ -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));
|
||||
}
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -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*/
|
||||
|
||||
Loading…
Reference in New Issue
Block a user