RxJava3中如何在并行流mapper方法获取主线程ThreadLocal变量？

你直接在mapper方法里调用data.get()会返回null，原因很简单：parallel(4)配合runOn(io())会把任务切换到RxJava IO调度器的线程池线程上，而ThreadLocal的变量是线程绑定的，主线程的ThreadLocal数据无法直接被其他线程读取。

下面是几种可行的实现方式：

1. 主线程预取值，通过流传递给mapper #

这是最直接且安全的方式，在主线程先取出ThreadLocal的值，再把它和流中的元素一起传递给mapper：

// 主线程提前取出ThreadLocal的值（Java 8+自动支持effectively final）
String threadLocalValue = data.get();

Flowable.fromIterable(xxx)
        .parallel(4)
        .runOn(io())
        // 把预取的值传给mapper，根据需求修改mapper的参数签名
        .map(item -> mapper(item, threadLocalValue))
        .sequential()
        .blockingStream().collect(Collectors.toList());

// 对应的mapper方法修改为接收两个参数
private YourResultType mapper(YourItemType item, String threadLocalData) {
    // 在这里直接使用threadLocalData即可
    // ...你的业务逻辑
}

如果不想修改mapper的参数签名，可以用简单的包装类把元素和ThreadLocal值打包：

String threadLocalValue = data.get();

Flowable.fromIterable(xxx)
        .map(item -> new AbstractMap.SimpleEntry<>(item, threadLocalValue))
        .parallel(4)
        .runOn(io())
        .map(entry -> {
            // 拆包后调用原mapper方法
            return mapper(entry.getKey());
        })
        .sequential()
        .blockingStream().collect(Collectors.toList());

// 原mapper方法可通过闭包访问threadLocalValue（变量需为effectively final）
private YourResultType mapper(YourItemType item) {
    // 直接使用threadLocalValue
    // ...
}

2. 利用RxJava的Context上下文传递（RxJava 3+） #

如果使用RxJava 3及以上版本，可以利用其内置的Context机制自动在线程间传递数据，无需手动打包元素：

Flowable.fromIterable(xxx)
        // 主线程将ThreadLocal值写入Context
        .contextWrite(Context.of("threadLocalData", data.get()))
        .parallel(4)
        .runOn(io())
        .map(item -> {
            // 在IO线程读取Context中的值
            String threadLocalValue = Context.current().get("threadLocalData");
            // 调用mapper并传入值
            return mapper(item, threadLocalValue);
        })
        .sequential()
        .blockingStream().collect(Collectors.toList());

• 如果ThreadLocal存储的是可变对象，要确保多线程访问时的线程安全（比如加锁、使用线程安全容器），避免多个IO线程同时修改导致数据混乱。
• 不要尝试在线程池线程中直接设置主线程的ThreadLocal，这会污染线程池的复用线程，导致后续任务拿到错误的数据。

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