在尝试OSGi PushStream库时,我觉得它确实很慢。我创建了两个方法,它们使用PushStream进行相同的操作,而另一个方法使用简单的BlockingQueue(请参见下面的代码),结果如下:

Queue needs 3 milliseconds to process 1000 events.
PushStream needs 31331 milliseconds to process 1000 events.


为什么PushStream慢一些?我做错了什么?



使用PushStream:

public class TestPush{

    @Test
    public void testPushStream() throws Exception {
        final PromiseFactory pf = new PromiseFactory(PromiseFactory.inlineExecutor());

        final PushStreamProvider psp = new PushStreamProvider();
        final SimplePushEventSource<Integer> source =
              psp.buildSimpleEventSource(Integer.class).withQueuePolicy(QueuePolicyOption.BLOCK).build();

        final Deferred<Instant> startD = pf.deferred();
        final Deferred<Instant> endD = pf.deferred();

        psp.createStream(source).onClose(() -> endD.resolve( Instant.now()) ).forEach((i) -> {
            if (i == 0) {
                startD.resolve( Instant.now() );
            }
        });

        final Promise<Long> nbEvent = psp.createStream(source).count();

        for (int i = 0; i < 1000; i++) {
            source.publish(i);
        }
        source.endOfStream();

        System.out.println("PushStream needs "
        + Duration.between( startD.getPromise().getValue(), endD.getPromise().getValue() ).toMillis()
        + " milliseconds to process " + nbEvent.getValue() + " events.");
    }


使用ArrayBlockingQueue:

    @Test
    public void testBlockingQueue() throws Exception {
        final PromiseFactory pf = new PromiseFactory(PromiseFactory.inlineExecutor());

        final Executor e = Executors.newFixedThreadPool(1);
        final ArrayBlockingQueue<Integer> abq = new ArrayBlockingQueue<>(32);

        final Deferred<Instant> startD = pf.deferred();
        final Deferred<Instant> endD = pf.deferred();
        final Deferred<Integer> nbEvent = pf.deferred();

        e.execute( () -> {
            try {
                Integer i = 0;
                Integer last = 0;
                do {
                    i = abq.take();

                    if (i == 0) {
                        startD.resolve(Instant.now());
                    } else if (i != -1) {
                        last = i;
                    }
                }
                while (i != -1);
                endD.resolve(Instant.now());
                nbEvent.resolve(last + 1);
            }
            catch (final InterruptedException exception) {
                exception.printStackTrace();
            }
        });

        for (int i = 0; i < 1000; i++) {
            abq.put(i);
        }
        abq.put(-1);

        System.out.println("Queue needs "
        + Duration.between( startD.getPromise().getValue(), endD.getPromise().getValue() ).toMillis()
        + " milliseconds to process " + nbEvent.getPromise().getValue() + " events.");
    }
}

最佳答案

这是一个有趣的问题:)


为什么PushStream慢一些?我做错了什么?


感谢您不仅仅假定PushStream实现很烂。在这种情况下,它会变慢,因为(可能没有意识到)您要求它是这样!

第1部分-缓冲

默认情况下,PushStreams被缓冲。这意味着它们包括一个队列,在处理事件之前将其放入其中。因此,缓冲会做一些不利于吞吐速度的事情。


它将额外的队列/出队步骤添加到管道中
它在事件处理中添加了一个额外的线程开关
缓冲区的默认策略是返回与缓冲区已满有关的压力。


在这种情况下,减速的绝大部分是由于背压。使用psp.createStream(source)创建流时,该流将设置有32个元素的缓冲区以及基于缓冲区大小的线性反压策略,满时返回一秒钟,如果其中有一个则返回31毫秒。值得注意的是,每个元素31毫秒总计30秒!

重要的是,SimplePushEventSource始终接受来自添加到它的使用者的反压力请求。这意味着您可能会尽可能快地将事件泵入SimplePushEventSource,但是它们只会按照管道请求的速度尽快传递。

如果我们从正在创建的推送流中删除缓冲,那么我们将进行以下测试:

@Test
public void testPushStream2() throws Exception {
    final PromiseFactory pf = new PromiseFactory(PromiseFactory.inlineExecutor());

    final PushStreamProvider psp = new PushStreamProvider();
    final SimplePushEventSource<Integer> source =
          psp.buildSimpleEventSource(Integer.class)
          .withQueuePolicy(QueuePolicyOption.BLOCK)
          .build();

    final Deferred<Instant> startD = pf.deferred();
    final Deferred<Instant> endD = pf.deferred();

    psp.buildStream(source).unbuffered().build().onClose(() -> endD.resolve( Instant.now()) ).forEach((i) -> {
        if (i == 0) {
            startD.resolve( Instant.now() );
        }
    });

    final Promise<Long> nbEvent = psp.buildStream(source).unbuffered().build().count();

    for (int i = 0; i < 1000; i++) {
        source.publish(i);
    }
    source.endOfStream();

    System.out.println("PushStream needs "
    + Duration.between( startD.getPromise().getValue(), endD.getPromise().getValue() ).toMillis()
    + " milliseconds to process " + nbEvent.getValue() + " events.");
}


(在我的机器上)运行此命令的结果是:

PushStream needs 39 milliseconds to process 1000 events.


这显然更接近您的期望,但仍然明显慢一些。请注意,我们可能仍然会有一些缓冲,但是请调整PushbackPolicy。这样可以为我们提供更快的吞吐率,但还不如我们这么快。

第2部分-管道长度

接下来要注意的是,您正在使用onClose()处理程序。这为推送流管道增加了一个额外的阶段。实际上,您可以将onClose移动为promise的结果,从而减少了管道的长度(您只需要运行一次)。

@Test
public void testPushStream3() throws Exception {
    final PromiseFactory pf = new PromiseFactory(PromiseFactory.inlineExecutor());

    final PushStreamProvider psp = new PushStreamProvider();
    final SimplePushEventSource<Integer> source =
            psp.buildSimpleEventSource(Integer.class)
            .withQueuePolicy(QueuePolicyOption.BLOCK)
            .build();

    final Deferred<Instant> startD = pf.deferred();
    final Deferred<Instant> endD = pf.deferred();

    psp.buildStream(source).unbuffered().build().forEach((i) -> {
        if (i == 0) {
            startD.resolve( Instant.now() );
        }
    });

    final Promise<Long> nbEvent = psp.buildStream(source).unbuffered().build().count()
            .onResolve(() -> endD.resolve( Instant.now()));

    for (int i = 0; i < 1000; i++) {
        source.publish(i);
    }
    source.endOfStream();

    System.out.println("PushStream needs "
            + Duration.between( startD.getPromise().getValue(), endD.getPromise().getValue() ).toMillis()
            + " milliseconds to process " + nbEvent.getValue() + " events.");
}


这个版本的结果(在我的机器上)是:

PushStream needs 21 milliseconds to process 1000 events.


第3部分-多路传输

“原始数组阻止队列”示例与PushStream示例之间的主要区别在于,您实际上创建了两个PushStream。第一个工作是捕获开始时间,第二个工作是计算事件。这将强制SimplePushEventSource在多个使用者上复用事件。

如果将行为折叠到单个管道中,以便SimplePushEventSource可以使用快速路径传递怎么办?

@Test
public void testPushStream4() throws Exception {
    final PromiseFactory pf = new PromiseFactory(PromiseFactory.inlineExecutor());

    final PushStreamProvider psp = new PushStreamProvider();
    final SimplePushEventSource<Integer> source =
            psp.buildSimpleEventSource(Integer.class)
            .withQueuePolicy(QueuePolicyOption.BLOCK)
            .build();

    final Deferred<Instant> startD = pf.deferred();
    final Deferred<Instant> endD = pf.deferred();

    final Promise<Long> nbEvent = psp.buildStream(source).unbuffered().build()
            .filter(i -> {
                if (i == 0) {
                    startD.resolve( Instant.now() );
                }
                return true;
            })
            .count()
            .onResolve(() -> endD.resolve( Instant.now()));

    for (int i = 0; i < 1000; i++) {
        source.publish(i);
    }
    source.endOfStream();

    System.out.println("PushStream needs "
            + Duration.between( startD.getPromise().getValue(), endD.getPromise().getValue() ).toMillis()
            + " milliseconds to process " + nbEvent.getValue() + " events.");
}


这个版本的结果(在我的机器上)是:

PushStream needs 3 milliseconds to process 1000 events.


摘要

PushStream是一种消耗异步到达事件的快速有效的方法,但是了解哪种缓冲行为适合您的应用程序非常重要。如果您想快速地遍历大量数据,则需要注意如何进行设置,因为缓冲默认值是为不同的用例设计的!

07-28 06:44