当前位置：首页 > 文章列表 > 文章 > java教程 > 带优先级的响应式生产者消费者模式实现

带优先级的响应式生产者消费者模式实现

2025-10-30 21:45:49 0浏览收藏

积累知识，胜过积蓄金银！毕竟在文章开发的过程中，会遇到各种各样的问题，往往都是一些细节知识点还没有掌握好而导致的，因此基础知识点的积累是很重要的。下面本文《响应式编程中实现带优先级和可控缓冲的生产者-消费者模式》，就带大家讲解一下知识点，若是你对本文感兴趣，或者是想搞懂其中某个知识点，就请你继续往下看吧~

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

在Java Reactor的生产者-消费者模式中，当内置Sinks无法满足任务优先级、队列监控及清空需求时，可利用`Sinks.many().unicast().onBackpressureBuffer()`结合外部`PriorityQueue`实现高效、可控的异步任务处理，避免阻塞式操作，从而构建一个功能更强大的响应式任务处理系统。

1. Reactor生产者-消费者模式中的挑战

在基于Reactor的应用程序中，生产者-消费者模式常用于异步任务处理。通常，我们会使用Sinks.Many来在生产者和消费者之间传递数据，例如：

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();

这种实现方式在大多数情况下运行良好。然而，当系统在高负载下运行时，我们可能会遇到以下痛点：

队列可见性： 无法直接获取Sink中当前待处理任务的数量。
队列清空： 无法方便地清空Sink中所有待处理任务。
任务优先级： 无法对Sink中的任务进行优先级排序。

由于标准Sinks.Many不提供对内部缓冲队列的直接访问，上述需求难以满足。

2. 避免阻塞式操作：为什么poll()在响应式编程中是问题

为了解决上述问题，一种常见的尝试是引入自定义的包装类，其中包含一个PriorityBlockingQueue，并通过Flux.create结合poll()方法从队列中获取元素：

// 自定义任务队列
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();

尽管这种方法似乎解决了优先级和队列访问的问题，但其中使用PriorityBlockingQueue.poll(1, TimeUnit.SECOND)是一个阻塞式操作。响应式编程的核心目标之一就是避免阻塞，因为阻塞操作会占用线程并等待，这与Reactor的非阻塞、异步特性相悖。在长时间运行测试中，阻塞式poll()可能导致线程挂起，严重影响系统的响应性和吞吐量。

3. Reactor原生解决方案：结合Sinks.many().unicast()与外部PriorityQueue

Reactor提供了一个更优雅、更符合响应式编程原则的解决方案：利用Sinks.many().unicast().onBackpressureBuffer(Queue)。这个方法允许我们提供一个外部的Queue作为Sink的背压缓冲区。通过使用PriorityQueue作为这个外部队列，我们可以完美地解决任务优先级、队列可见性和清空的问题，同时保持非阻塞特性。

3.1 核心原理

Sinks.many().unicast().onBackpressureBuffer(Queue)的unicast特性意味着只有一个订阅者可以消费Sink发出的元素。当这个订阅者无法及时处理元素时，新发出的元素会被存储到我们提供的外部Queue中。

通过这种方式，我们可以：

实现优先级： 将一个PriorityQueue作为外部队列，并定义好任务的比较器，Sink将自动按照优先级从队列中取出任务。
监控队列： 直接访问外部PriorityQueue实例，调用其size()方法即可获取当前待处理任务数量。
清空队列： 直接调用外部PriorityQueue实例的clear()方法即可清空所有待处理任务。

3.2 示例代码

下面是一个演示如何使用外部PriorityQueue与Sinks.many().unicast()实现带优先级、可监控的生产者-消费者模式的例子。

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();
    }
}

3.3 运行输出分析

运行上述代码，你可能会看到类似以下的输出（时间戳会有所不同）：

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: 演示结束.

输出解释：

当前Sink中待处理任务数量: 9: 在所有任务发射完毕后，由于消费者有100ms的延迟，大部分任务都进入了taskQueue。第一个任务Task-0可能在生产者循环结束前就被delayElements的内部队列捕获并开始处理，所以外部队列中剩下9个。
处理任务: Task[prio=0, name=Task-0]: 尽管PriorityQueue通常会先处理优先级最高的任务，但delayElements操作符本身有一个内部队列（通常大小为1）。这意味着当Task-0被Sink发出后，它可能立即进入delayElements的内部队列并开始计时，在Task-1被发出之前就已经被消费者处理。
处理任务: Task[prio=9, name=Task-9]: 紧接着，PriorityQueue的优先级特性开始生效。Task-9（优先级最高）被取出并处理。
处理任务: Task[prio=8, name=Task-8]: 随后是Task-8。
清空后Sink中待处理任务数量: 0: 调用taskQueue.clear()后，外部队列被清空。
处理任务: Task[prio=7, name=Task-7]: 尽管外部队列已清空，但Task-7仍然被处理了。这是因为在taskQueue.clear()被调用之前，Task-7可能已经从taskQueue中取出，并进入了delayElements操作符的内部队列中等待处理。

4. 关于多播（Multicast）的需求

上述解决方案使用了unicast Sink，这意味着只有一个订阅者可以消费其发出的元素。如果您的业务场景确实需要多个消费者订阅同一个Flux，并让他们都能接收到外部PriorityQueue中按优先级取出的任务，您可以在taskSink.asFlux()之后，利用Reactor提供的多播操作符来实现：

// 如果需要多播，可以在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); // 第二个消费者

重要提示： 在这种多播场景下，虽然外部PriorityQueue确保了任务在进入Sink时的优先级，但一旦任务被Sink发出并进入多播管道，每个订阅者会独立接收到这些任务。如果多个消费者需要独立地、按照自己的节奏处理任务，且每个消费者都需要完整的优先级队列功能，那么可能需要为每个消费者维护一个独立的unicast Sink和PriorityQueue，或者重新评估多播的必要性。