Reactor响应式编程中如何实现带优先级和可控缓冲的生产者-消费者模式

Sinks.Many<Task> taskSink = Sinks.many().multicast().onBackpressureBuffer(1000, false);

// 生产者
Flux<Date> dates = loadDates();
dates.filterWhen(...)
     .concatMap(date -> taskManager.getTaskByDate(date))
     .doOnNext(taskSink::tryEmitNext)
     .subscribe();

// 消费者
taskProcessor.process(taskSink.asFlux())
             .subscribeOn(Schedulers.boundedElastic())
             .subscribe();

// 自定义任务队列
MergingQueue<Task> taskQueue = new PriorityMergingQueue();

// 生产者
Flux<Date> dates = loadDates();
dates.filterWhen(...)
     .concatMap(date -> taskManager.getTaskByDate(date))
     .doOnNext(taskQueue::enqueue) // 将任务加入自定义队列
     .subscribe();

// 消费者
taskProcessor.process(Flux.create((sink) -> {
     sink.onRequest(n -> {
          Task task;
          try {
                while(!sink.isCancel() && n > 0) {
                    // 潜在的阻塞操作
                    if((task = taskQueue.poll(1, TimeUnit.SECOND))  != null) {
                        sink.next(task);
                        n--;
                    }
                }
          } catch(InterruptedException e) {
                Thread.currentThread().interrupt();
                sink.error(e);
          }
     });
}))
.subscribeOn(Schedulers.boundedElastic())
.subscribe();

import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.time.LocalTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.Comparator;
import java.util.Queue;
import java.util.PriorityQueue;
import java.util.concurrent.TimeUnit;

// 任务记录类，包含优先级
record Task(int prio, String name) {}

public class PriorityTaskProcessor {

    private static void log(Object message) {
        System.out.println(LocalTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.MILLIS) + ": " + message);
    }

    public void externalBufferDemo() throws InterruptedException {
        // 1. 创建一个PriorityQueue作为外部缓冲区
        // 优先级高的（prio值大）的任务先处理，所以使用reversed()
        Queue<Task> taskQueue = new PriorityQueue<>(Comparator.comparingInt(Task::prio).reversed());

        // 2. 创建unicast Sink，并指定使用外部的PriorityQueue作为背压缓冲区
        Sinks.Many<Task> taskSink = Sinks.many().unicast().onBackpressureBuffer(taskQueue);

        // 3. 消费者：订阅Sink发出的Flux
        // 为了演示效果，这里模拟一个处理延迟
        taskSink.asFlux()
                .delayElements(Duration.ofMillis(100)) // 模拟每个任务处理需要100ms
                .doOnNext(task -> log("处理任务: " + task))
                .subscribeOn(Schedulers.boundedElastic()) // 在弹性调度器上执行处理逻辑
                .subscribe(
                        task -> {}, // onNext
                        error -> log("消费者发生错误: " + error.getMessage()), // onError
                        () -> log("消费者完成") // onComplete
                );

        // 4. 生产者：向Sink发射任务
        log("开始发射任务...");
        for (int i = 0; i < 10; i++) {
            // 发射不同优先级的任务
            taskSink.tryEmitNext(new Task(i, "Task-" + i));
            // 模拟生产者快速生产
            Thread.sleep(10);
        }
        log("任务发射完毕.");

        // 5. 检查Sink中任务数量（直接访问外部队列）
        log("当前Sink中待处理任务数量: " + taskQueue.size());

        // 6. 模拟一段时间后清空队列
        Thread.sleep(350); // 等待一些任务被处理
        log("准备清空Sink中所有待处理任务...");
        taskQueue.clear(); // 直接清空外部PriorityQueue
        log("清空后Sink中待处理任务数量: " + taskQueue.size());

        // 7. 继续等待，观察清空后的处理情况
        Thread.sleep(1500);
        log("演示结束.");
    }

    public static void main(String[] args) throws InterruptedException {
        new PriorityTaskProcessor().externalBufferDemo();
    }
}

09:41:11.347: 开始发射任务...
09:41:11.437: 任务发射完毕.
09:41:11.437: 当前Sink中待处理任务数量: 9
09:41:11.539: 处理任务: Task[prio=0, name=Task-0] // Task-0先被处理，因为delayElements的内部队列
09:41:11.642: 处理任务: Task[prio=9, name=Task-9] // 队列中的最高优先级任务
09:41:11.745: 处理任务: Task[prio=8, name=Task-8] // 接下来是优先级8的任务
09:41:11.787: 准备清空Sink中所有待处理任务...
09:41:11.787: 清空后Sink中待处理任务数量: 0
09:41:11.848: 处理任务: Task[prio=7, name=Task-7] // 注意：此任务在清空后仍被处理
09:41:12.051: 演示结束.

// 如果需要多播，可以在unicast Sink的Flux上应用多播操作符
taskSink.asFlux()
        .publish() // 或 .share(), .replay() 等
        .autoConnect(2) // 示例：等待2个订阅者连接后开始发射
        .delayElements(Duration.ofMillis(100))
        .subscribe(consumer1); // 第一个消费者
taskSink.asFlux()
        .publish() // 再次强调，多播操作符应作用在原始Flux上，而不是创建多个Flux
        .autoConnect(2)
        .delayElements(Duration.ofMillis(100))
        .subscribe(consumer2); // 第二个消费者