java - 如何在 Java 和 logback 中将 MDC 与 parallelStream 一起使用

Question

我需要记录请求的一些属性，例如请求 ID 和语言环境，但是在使用 parallelStream 时，MDC 的 ThreadLocal 似乎丢失了这些信息。

我分析过创建parallelStream时在线程之间传递MDC上下文的解决方案，但是看起来很脏，而且我也有很多parallelStream的用法。

还有其他方法吗？

谢谢

Answer 1

我找到的唯一解决方案是将上下文复制到流外的最终变量中，并将其应用于每次迭代:

Map<String, String> contextMap = MDC.getCopyOfContextMap();
Stream.iterate(0, i -> i + 1).parallel()
    .peek(i -> MDC.setContextMap(contextMap))
    // ...logic...
    // in case you're using a filter, you need to use a predicate and combine it with a clear step:
    filter(yourPredicate.or(i -> {
                MDC.clear();
                return false;
            }))
    // clear right before terminal operation
    .peek(i -> MDC.clear())
    .findFirst();

// since the initial thread is also used within the stream and the context is cleared there, 
// we need to set it again to its initial state
MDC.setContextMap(contextMap);    

该解决方案的成本是 1) 每 100 次迭代几微秒和 2) 可读性较差，但两者都是可以接受的:

1. 这是将 IntStream.range(0, 100).parallel().sum()(=baseline)与使用该 MDC 复制逻辑的相同流进行比较的基准:

Benchmark               Mode  Cnt   Score   Error   Units
MDC_CopyTest.baseline  thrpt    5   0,038 ± 0,005  ops/us
MDC_CopyTest.withMdc   thrpt    5   0,024 ± 0,001  ops/us
MDC_CopyTest.baseline   avgt    5  28,239 ± 1,308   us/op
MDC_CopyTest.withMdc    avgt    5  40,178 ± 0,761   us/op

1. 为了提高可读性，可以将其包装到一个小的辅助类中:

public class MDCCopyHelper {
    private Map<String, String> contextMap = MDC.getCopyOfContextMap();

    public void set(Object... any) {
        MDC.setContextMap(contextMap);
    }

    public void clear(Object... any) {
        MDC.clear();
    }

    public boolean clearAndFail() {
        MDC.clear();
        return false;
    }
}

流式代码看起来更漂亮:

MDCCopyHelper mdcHelper = new MDCCopyHelper();
try {
    Optional<Integer> findFirst = Stream.iterate(0, i -> i + 1)
        .parallel()
        .peek(mdcHelper::set)
        // ...logic...
        // filters predicates should be combined with clear step
        .filter(yourPredicate.or(mdcHelper::clearAndFail))
        // before terminal call:
        .peek(mdcHelper::clear)
        .findFirst();
} finally {
    // set the correct MDC at the main thread again
    mdcHelper.set();
}