了解 akka 流 Source.queue 中的背压

Question

我正在尝试 akka 流，但在我的简单示例中我无法获得背压工作。诚然，我对 akka（流）没有经验，所以我可能遗漏了一些重要的东西。

我产生（在队列上提供）整数比消耗它们更快，所以我认为背压会起作用。 我的目标是始终使用放入队列中的最新项目（这就是为什么我有 bufferSize = 1 和 OverflowStrategy.dropHead() 在源队列上）。

public class SimpleStream {
    public static void main(String[] argv) throws InterruptedException {
        final ActorSystem system = ActorSystem.create("akka-streams");
        final Materializer materializer = ActorMaterializer.create(system);

        final Procedure<Integer> slowConsumer = (i) -> {
            System.out.println("consuming [" + i + "]");
            ThreadUtils.sleepQuietly(1000, TimeUnit.MILLISECONDS);
        };

        final SourceQueue<Integer> q = Sink
                .<Integer>foreach(slowConsumer)
                .runWith(Source.<Integer>queue(1, OverflowStrategy.dropHead()),  materializer);

        final AtomicInteger i = new AtomicInteger(0);
        final Thread t = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                int n = i.incrementAndGet(); 
                q.offer(n);
                System.out.println("produced: [" + n + "]");
                ThreadUtils.sleepQuietly(500, TimeUnit.MILLISECONDS);
            }
        });
        t.setName("ticking");
        t.start();

        // run some time... to observe the effects.
        ThreadUtils.sleepQuietly(1, TimeUnit.HOURS);
        t.interrupt();
        t.join();

        // eventually shutdown akka here...
    }
}

但是结果是这样的：

produced: [1]
consuming [1]
produced: [2]
produced: [3]
consuming [2] <-- Expected to be consuming 3 here.
produced: [4]
produced: [5]
consuming [3] <-- Expected to be consuming 5 here.
produced: [6]
produced: [7]

请忽略这里和那里的线程，只是为了伪造从外部获取数据 源（如果我必须在实际项目中使用它会发生这种情况）。

知道我缺少什么吗？

Answer 1

Source.queue 终止背压信号。这就是 Source.queue 方法接受OverflowStrategy 的原因。如果可以通过队列向上游发出背压信号，则无需处理队列可能溢出的情况。但由于背压不会通过队列传播，因此需要定义策略来处理比消费者更快的生产者。

对于典型的流，终极 Source 接收来自 Sink 的需求以产生更多结果。但是，对于从Source.queue 创建的流，“最终源”是一个队列。这个队列只能排出内容，如果有的话。它无法向上游发出信号以生成更多结果，因为上游位于 offer 方法的另一侧。