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