Java EE环境下如何共享ContainerRequestFilter生成的RabbitMQ处理UID

1. 完善请求拦截器：生成并管理UID，同时清理资源 #

首先扩展你的拦截器，添加请求结束后的清理逻辑（避免ThreadLocal内存泄漏），并提供UID的获取入口：

@Provider
@PreMatching
@LogHttp
public class LogInterceptor implements ContainerRequestFilter, ContainerResponseFilter {

    private static final ThreadLocal<String> REQUEST_UID = new ThreadLocal<>();
    // 注入RabbitMQ日志队列生产者（假设已实现）
    @Inject
    private RabbitMqLogProducer logProducer;

    @Override
    public void filter(ContainerRequestContext requestContext) throws IOException {
        // 生成唯一UID
        String uid = UUID.randomUUID().toString();
        REQUEST_UID.set(uid);

        // 组装日志信息并发送到日志队列
        String logContent = String.format("Request received | UID: %s | Method: %s | Path: %s",
                uid,
                requestContext.getMethod(),
                requestContext.getUriInfo().getPath());
        logProducer.sendToLogQueue(logContent);
    }

    @Override
    public void filter(ContainerRequestContext requestContext, ContainerResponseContext responseContext) throws IOException {
        // 请求结束后强制清理ThreadLocal，防止内存泄漏
        REQUEST_UID.remove();
    }

    // 提供静态方法供其他组件获取当前请求的UID
    public static String getCurrentUid() {
        return REQUEST_UID.get();
    }
}

2. 在REST资源中传递UID到业务消息 #

在处理请求的REST接口里，获取UID并将其作为消息属性附加到RabbitMQ业务消息中，实现日志与消息处理的关联：

@Path("/data")
@Consumes(MediaType.APPLICATION_JSON)
public class DataResource {

    @Inject
    private RabbitMqBusinessProducer businessProducer;

    @POST
    public Response submitData(DataDto data) {
        // 获取拦截器生成的UID
        String requestUid = LogInterceptor.getCurrentUid();

        // 将UID作为RabbitMQ消息的Header传递
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                .headers(Map.of("REQUEST_UID", requestUid))
                .build();

        // 发送业务消息到队列
        businessProducer.sendToBusinessQueue(data, props);

        return Response.accepted().entity(Map.of("requestUid", requestUid)).build();
    }
}

3. 在消息消费者中关联UID与处理日志 #

消费业务消息时，从消息Header中取出UID，存入ThreadLocal后再进行处理，确保日志能关联到原始请求：

public class DataConsumer implements Consumer {

    private static final Logger LOGGER = Logger.getLogger(DataConsumer.class.getName());
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        // 从消息Header中提取UID
        String requestUid = (String) properties.getHeaders().get("REQUEST_UID");
        // 存入ThreadLocal，方便后续处理日志使用
        LogInterceptor.REQUEST_UID.set(requestUid);

        try {
            DataDto data = objectMapper.readValue(body, DataDto.class);
            // 日志中携带UID，实现请求与处理链路的关联
            LOGGER.info("Processing data | UID: %s | Content: %s".formatted(requestUid, data.getContent()));

            // 业务处理逻辑...

        } finally {
            // 处理完成后清理ThreadLocal
            LogInterceptor.REQUEST_UID.remove();
        }
    }

    // 实现Consumer接口的其他方法（如handleConsumeOk、handleCancel等）
}

关键注意事项 #

• ThreadLocal清理：必须在请求结束（ResponseFilter）和消息消费完成后执行清理，避免线程池中的线程持有旧的UID导致数据混乱。
• 异步场景适配：RabbitMQ消费属于异步处理，无法直接复用请求线程的ThreadLocal，因此必须通过消息Header传递UID，再在消费线程中临时存入ThreadLocal。

内容的提问来源于stack exchange，提问作者OceanWavez