Reactor Sink/Processor选型：线程隔离的重复消息检测实现

首先得给你点个赞，已经精准抓住了问题的核心——既然消息已经通过JMS分组/Kinesis分片做了预分组，同一个线程处理的消息绝不会和其他线程的重复，那之前把所有线程的消息塞进同一个EmitterProcessor的做法确实是浪费，而且完全没必要混排不同线程的消息。

我们的核心思路很简单：给每个调用onMessage的线程分配一个专属的消息流，让每个线程的消息在独立的流里做重复检测，彻底避免跨线程的消息混排。这里用ThreadLocal来绑定每个线程的Processor是最直接的方案，配合Reactor的轻量处理器UnicastProcessor就能完美实现。

优化后的完整实现代码 #

import reactor.core.publisher.Flux;
import reactor.core.publisher.UnicastProcessor;

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;

public class ThreadIsolatedConsumer implements Consumer {
    // 用ThreadLocal维护每个线程专属的Processor
    private final ThreadLocal<UnicastProcessor<Object>> threadLocalProcessor;
    // 存储每个Processor的订阅状态，避免重复订阅
    private final Map<UnicastProcessor<Object>, Boolean> subscribedProcessors = new ConcurrentHashMap<>();
    private final int maxHits;
    private final int bufferSize;
    private final Duration bufferTimeout;

    public ThreadIsolatedConsumer(int maxHits, int bufferSize, long timeoutMillis) {
        this.maxHits = maxHits;
        this.bufferSize = bufferSize;
        this.bufferTimeout = Duration.ofMillis(timeoutMillis);

        // 初始化ThreadLocal，每个线程首次调用时创建对应的Processor
        this.threadLocalProcessor = ThreadLocal.withInitial(() -> {
            UnicastProcessor<Object> processor = UnicastProcessor.create();
            // 为新创建的Processor订阅处理逻辑（仅执行一次）
            subscribedProcessors.computeIfAbsent(processor, p -> {
                p.bufferTimeout(bufferSize, bufferTimeout)
                  .flatMapIterable(Function.identity()) // 将buffer展开为单个元素
                  .transform(values -> getExceedingRates(values, maxHits, bufferSize, bufferTimeout))
                  .subscribe(cache -> {
                      System.out.println("线程[" + Thread.currentThread().getName() + "]检测到重复消息: " + cache.getExceedingValues());
                  });
                return true;
            });
            return processor;
        });
    }

    @Override
    public void onMessage(Object message) {
        // 获取当前线程的Processor，发送消息
        threadLocalProcessor.get().onNext(message);
    }

    // 你的重复检测方法保持不变
    public static <T> Flux<OcurrenceCache<T>> getExceedingRates(Flux<T> values, int maxHits, int bufferSize, Duration bufferTimeout) {
        return values.bufferTimeout(bufferSize, bufferTimeout)
                .map(vals -> {
                    OcurrenceCache<T> occurrenceCache = new OcurrenceCache<>(maxHits);
                    for (T value : vals) {
                        occurrenceCache.incrementNrOccurrences(value);
                    }
                    return occurrenceCache;
                });
    }

    // 可选：提供资源清理方法，在线程池销毁时调用，避免内存泄漏
    public void dispose() {
        subscribedProcessors.keySet().forEach(processor -> {
            processor.onComplete();
            processor.dispose();
        });
        subscribedProcessors.clear();
        threadLocalProcessor.remove();
    }
}

// 假设你的OcurrenceCache类实现如下（保持你的原有逻辑）
class OcurrenceCache<T> {
    private final int maxHits;
    private final Map<T, Integer> occurrenceMap = new ConcurrentHashMap<>();

    public OcurrenceCache(int maxHits) {
        this.maxHits = maxHits;
    }

    public void incrementNrOccurrences(T value) {
        occurrenceMap.compute(value, (k, v) -> v == null ? 1 : v + 1);
    }

    public Map<T, Integer> getExceedingValues() {
        return occurrenceMap.entrySet().stream()
                .filter(entry -> entry.getValue() > maxHits)
                .collect(Map::of, (m, e) -> m.put(e.getKey(), e.getValue()), Map::putAll);
    }
}

关键优化点解释 #

• 线程隔离的消息流：
  用ThreadLocal.withInitial为每个线程创建独立的UnicastProcessor，每个线程的onMessage调用只会把消息发送到自己的Processor里，完全不会和其他线程的消息混排，完美契合你的业务前提。

• 避免重复订阅：
  用ConcurrentHashMap记录每个Processor的订阅状态，确保每个Processor只会被订阅一次，不会因为线程多次调用onMessage而重复创建订阅逻辑。

• 资源安全：
  提供了dispose方法，当你的线程池关闭或者消费者不再使用时，可以调用这个方法清理所有Processor，避免内存泄漏。

• 简化处理逻辑：
  不再需要在subscribe里嵌套调用getExceedingRates，而是直接用transform操作符把重复检测逻辑整合到每个线程的流里，代码更清晰，可读性更强。

这个方案既满足了“单次onMessage调用无副作用”的要求，又最大化利用了你的预分组优势，性能和逻辑清晰度都比之前的实现好很多。

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