1 测试环境
2 OneToOneSequencedThroughputTest 测试报告
分别使用四种等待策略，两种写入模式，进行测试：
/**
* <pre>
* UniCast a series of items between 1 publisher and 1 event processor.
*
* +----+ +-----+
* | P1 |--->| EP1 |
* +----+ +-----+
*
* Disruptor:
* ==========
* track to prevent wrap
* +------------------+
* | |
* | v
* +----+ +====+ +====+ +-----+
* | P1 |--->| RB |<---| SB | | EP1 |
* +----+ +====+ +====+ +-----+
* claim get ^ |
* | |
* +--------+
* waitFor
*
* P1 - Publisher 1
* RB - RingBuffer
* SB - SequenceBarrier
* EP1 - EventProcessor 1
*
* </pre>
*/
public final class OneToOneSequencedThroughputTest extends AbstractPerfTestDisruptor {
private static final int BUFFER_SIZE = 1024 * 64;
private static final long ITERATIONS = 1000L * 1000L * 100L;
private final ExecutorService executor = Executors.newSingleThreadExecutor(DaemonThreadFactory.INSTANCE);
private final long expectedResult = PerfTestUtil.accumulatedAddition(ITERATIONS);
private RingBuffer<ValueEvent> ringBuffer;
private final SequenceBarrier sequenceBarrier;
private final ValueAdditionEventHandler handler;
private final BatchEventProcessor<ValueEvent> batchEventProcessor;
public OneToOneSequencedThroughputTest(ProducerType type, WaitStrategy strategy) {
if (type.equals(ProducerType.SINGLE)) {
this.ringBuffer = createSingleProducer(ValueEvent.EVENT_FACTORY, BUFFER_SIZE, strategy);
}
if (type.equals(ProducerType.MULTI)) {
this.ringBuffer = createMultiProducer(ValueEvent.EVENT_FACTORY, BUFFER_SIZE, strategy);
}
sequenceBarrier = ringBuffer.newBarrier();
handler = new ValueAdditionEventHandler();
batchEventProcessor = new BatchEventProcessor<ValueEvent>(ringBuffer, sequenceBarrier, handler);
ringBuffer.addGatingSequences(batchEventProcessor.getSequence());
}
@Override
protected int getRequiredProcessorCount() {
return 2;
}
@Override
protected long runDisruptorPass() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
long expectedCount = batchEventProcessor.getSequence().get() + ITERATIONS;
handler.reset(latch, expectedCount);
executor.submit(batchEventProcessor);
long start = System.currentTimeMillis();
final RingBuffer<ValueEvent> rb = ringBuffer;
for (long i = 0; i < ITERATIONS; i++) {
long next = rb.next();
rb.get(next).setValue(i);
rb.publish(next);
}
latch.await();
long opsPerSecond = (ITERATIONS * 1000L) / (System.currentTimeMillis() - start);
waitForEventProcessorSequence(expectedCount);
batchEventProcessor.halt();
failIfNot(expectedResult, handler.getValue());
return opsPerSecond;
}
private void waitForEventProcessorSequence(long expectedCount) throws InterruptedException {
while (batchEventProcessor.getSequence().get() != expectedCount) {
Thread.sleep(1);
}
}
public static void main(String[] args) throws Exception {
System.out.println("ProducerType.SINGLE, use yield wait strategy");
OneToOneSequencedThroughputTest test1 = new OneToOneSequencedThroughputTest(ProducerType.SINGLE, new YieldingWaitStrategy());
test1.testImplementations();
System.out.println("ProducerType.SINGLE, use sleep wait strategy");
OneToOneSequencedThroughputTest test2 = new OneToOneSequencedThroughputTest(ProducerType.SINGLE, new SleepingWaitStrategy());
test2.testImplementations();
System.out.println("ProducerType.SINGLE, use block wait strategy");
OneToOneSequencedThroughputTest test3 = new OneToOneSequencedThroughputTest(ProducerType.SINGLE, new BlockingWaitStrategy());
test3.testImplementations();
System.out.println("ProducerType.SINGLE, use busy spin wait strategy");
OneToOneSequencedThroughputTest test4 = new OneToOneSequencedThroughputTest(ProducerType.SINGLE, new BusySpinWaitStrategy());
test4.testImplementations();
System.out.println("ProducerType.MULTI, use yield wait strategy");
OneToOneSequencedThroughputTest test5 = new OneToOneSequencedThroughputTest(ProducerType.MULTI, new YieldingWaitStrategy());
test5.testImplementations();
System.out.println("ProducerType.MULTI, use sleep wait strategy");
OneToOneSequencedThroughputTest test6 = new OneToOneSequencedThroughputTest(ProducerType.MULTI, new SleepingWaitStrategy());
test6.testImplementations();
System.out.println("ProducerType.MULTI, use block wait strategy");
OneToOneSequencedThroughputTest test7 = new OneToOneSequencedThroughputTest(ProducerType.MULTI, new BlockingWaitStrategy());
test7.testImplementations();
System.out.println("ProducerType.MULTI, use busy spin wait strategy");
OneToOneSequencedThroughputTest test8 = new OneToOneSequencedThroughputTest(ProducerType.MULTI, new BusySpinWaitStrategy());
test8.testImplementations();
}
}
可以看到在ProducerType.SINGLE, use sleep wait strategy模式下性能最好。
ProducerType.SINGLE, use yield wait strategy
Starting Disruptor tests
Run 0, Disruptor=16,857,720 ops/sec
Run 1, Disruptor=16,829,350 ops/sec
Run 2, Disruptor=17,050,298 ops/sec
Run 3, Disruptor=16,526,194 ops/sec
Run 4, Disruptor=16,545,334 ops/sec
Run 5, Disruptor=16,477,179 ops/sec
Run 6, Disruptor=16,501,650 ops/sec
ProducerType.SINGLE, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=17,391,304 ops/sec
Run 1, Disruptor=17,562,346 ops/sec
Run 2, Disruptor=18,083,182 ops/sec
Run 3, Disruptor=17,730,496 ops/sec
Run 4, Disruptor=17,199,862 ops/sec
Run 5, Disruptor=17,280,110 ops/sec
Run 6, Disruptor=17,211,703 ops/sec
ProducerType.SINGLE, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=6,640,547 ops/sec
Run 1, Disruptor=6,676,904 ops/sec
Run 2, Disruptor=6,634,379 ops/sec
Run 3, Disruptor=6,671,114 ops/sec
Run 4, Disruptor=6,707,807 ops/sec
Run 5, Disruptor=6,498,992 ops/sec
Run 6, Disruptor=6,538,084 ops/sec
ProducerType.SINGLE, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=15,673,981 ops/sec
Run 1, Disruptor=15,647,003 ops/sec
Run 2, Disruptor=15,865,460 ops/sec
Run 3, Disruptor=15,908,367 ops/sec
Run 4, Disruptor=15,880,578 ops/sec
Run 5, Disruptor=15,850,372 ops/sec
Run 6, Disruptor=15,762,925 ops/sec
ProducerType.MULTI, use yield wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,934,937 ops/sec
Run 1, Disruptor=12,878,300 ops/sec
Run 2, Disruptor=12,965,123 ops/sec
Run 3, Disruptor=12,858,428 ops/sec
Run 4, Disruptor=12,873,326 ops/sec
Run 5, Disruptor=12,845,215 ops/sec
Run 6, Disruptor=12,805,736 ops/sec
ProducerType.MULTI, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,827,090 ops/sec
Run 1, Disruptor=12,953,367 ops/sec
Run 2, Disruptor=12,904,890 ops/sec
Run 3, Disruptor=12,931,591 ops/sec
Run 4, Disruptor=12,810,658 ops/sec
Run 5, Disruptor=13,118,194 ops/sec
Run 6, Disruptor=12,825,445 ops/sec
ProducerType.MULTI, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=8,179,290 ops/sec
Run 1, Disruptor=6,804,109 ops/sec
Run 2, Disruptor=6,862,005 ops/sec
Run 3, Disruptor=6,818,956 ops/sec
Run 4, Disruptor=6,824,541 ops/sec
Run 5, Disruptor=6,600,224 ops/sec
Run 6, Disruptor=6,650,262 ops/sec
ProducerType.MULTI, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,565,971 ops/sec
Run 1, Disruptor=12,591,286 ops/sec
Run 2, Disruptor=12,714,558 ops/sec
Run 3, Disruptor=12,690,355 ops/sec
Run 4, Disruptor=12,645,422 ops/sec
Run 5, Disruptor=12,701,638 ops/sec
Run 6, Disruptor=12,724,265 ops/sec
Starting Disruptor tests
Run 0, Disruptor=16,857,720 ops/sec
Run 1, Disruptor=16,829,350 ops/sec
Run 2, Disruptor=17,050,298 ops/sec
Run 3, Disruptor=16,526,194 ops/sec
Run 4, Disruptor=16,545,334 ops/sec
Run 5, Disruptor=16,477,179 ops/sec
Run 6, Disruptor=16,501,650 ops/sec
ProducerType.SINGLE, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=17,391,304 ops/sec
Run 1, Disruptor=17,562,346 ops/sec
Run 2, Disruptor=18,083,182 ops/sec
Run 3, Disruptor=17,730,496 ops/sec
Run 4, Disruptor=17,199,862 ops/sec
Run 5, Disruptor=17,280,110 ops/sec
Run 6, Disruptor=17,211,703 ops/sec
ProducerType.SINGLE, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=6,640,547 ops/sec
Run 1, Disruptor=6,676,904 ops/sec
Run 2, Disruptor=6,634,379 ops/sec
Run 3, Disruptor=6,671,114 ops/sec
Run 4, Disruptor=6,707,807 ops/sec
Run 5, Disruptor=6,498,992 ops/sec
Run 6, Disruptor=6,538,084 ops/sec
ProducerType.SINGLE, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=15,673,981 ops/sec
Run 1, Disruptor=15,647,003 ops/sec
Run 2, Disruptor=15,865,460 ops/sec
Run 3, Disruptor=15,908,367 ops/sec
Run 4, Disruptor=15,880,578 ops/sec
Run 5, Disruptor=15,850,372 ops/sec
Run 6, Disruptor=15,762,925 ops/sec
ProducerType.MULTI, use yield wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,934,937 ops/sec
Run 1, Disruptor=12,878,300 ops/sec
Run 2, Disruptor=12,965,123 ops/sec
Run 3, Disruptor=12,858,428 ops/sec
Run 4, Disruptor=12,873,326 ops/sec
Run 5, Disruptor=12,845,215 ops/sec
Run 6, Disruptor=12,805,736 ops/sec
ProducerType.MULTI, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,827,090 ops/sec
Run 1, Disruptor=12,953,367 ops/sec
Run 2, Disruptor=12,904,890 ops/sec
Run 3, Disruptor=12,931,591 ops/sec
Run 4, Disruptor=12,810,658 ops/sec
Run 5, Disruptor=13,118,194 ops/sec
Run 6, Disruptor=12,825,445 ops/sec
ProducerType.MULTI, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=8,179,290 ops/sec
Run 1, Disruptor=6,804,109 ops/sec
Run 2, Disruptor=6,862,005 ops/sec
Run 3, Disruptor=6,818,956 ops/sec
Run 4, Disruptor=6,824,541 ops/sec
Run 5, Disruptor=6,600,224 ops/sec
Run 6, Disruptor=6,650,262 ops/sec
ProducerType.MULTI, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=12,565,971 ops/sec
Run 1, Disruptor=12,591,286 ops/sec
Run 2, Disruptor=12,714,558 ops/sec
Run 3, Disruptor=12,690,355 ops/sec
Run 4, Disruptor=12,645,422 ops/sec
Run 5, Disruptor=12,701,638 ops/sec
Run 6, Disruptor=12,724,265 ops/sec
3 OneToOneSequencedBatchThroughputTest测试报告
每次批量写入10个对象，分别使用四种等待策略，两种写入模式，进行测试：/**
* <pre>
* UniCast a series of items between 1 publisher and 1 event processor.
*
* +----+ +-----+
* | P1 |--->| EP1 |
* +----+ +-----+
*
* Disruptor:
* ==========
* track to prevent wrap
* +------------------+
* | |
* | v
* +----+ +====+ +====+ +-----+
* | P1 |--->| RB |<---| SB | | EP1 |
* +----+ +====+ +====+ +-----+
* claim get ^ |
* | |
* +--------+
* waitFor
*
* P1 - Publisher 1
* RB - RingBuffer
* SB - SequenceBarrier
* EP1 - EventProcessor 1
*
* </pre>
*/
public final class OneToOneSequencedBatchThroughputTest extends AbstractPerfTestDisruptor {
public static final int BATCH_SIZE = 10;
private static final int BUFFER_SIZE = 1024 * 64;
private static final long ITERATIONS = 1000L * 1000L * 100L;
private final ExecutorService executor = Executors.newSingleThreadExecutor(DaemonThreadFactory.INSTANCE);
private final long expectedResult = PerfTestUtil.accumulatedAddition(ITERATIONS) * BATCH_SIZE;
private RingBuffer<ValueEvent> ringBuffer;
private final SequenceBarrier sequenceBarrier;
private final ValueAdditionEventHandler handler;
private final BatchEventProcessor<ValueEvent> batchEventProcessor;
public OneToOneSequencedBatchThroughputTest(ProducerType type, WaitStrategy strategy) {
if (type.equals(ProducerType.SINGLE)) {
this.ringBuffer = createSingleProducer(ValueEvent.EVENT_FACTORY, BUFFER_SIZE, strategy);
}
if (type.equals(ProducerType.MULTI)) {
this.ringBuffer = createMultiProducer(ValueEvent.EVENT_FACTORY, BUFFER_SIZE, strategy);
}
sequenceBarrier = ringBuffer.newBarrier();
handler = new ValueAdditionEventHandler();
batchEventProcessor = new BatchEventProcessor<ValueEvent>(ringBuffer, sequenceBarrier, handler);
ringBuffer.addGatingSequences(batchEventProcessor.getSequence());
}
@Override
protected int getRequiredProcessorCount() {
return 2;
}
@Override
protected long runDisruptorPass() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
long expectedCount = batchEventProcessor.getSequence().get() + ITERATIONS * BATCH_SIZE;
handler.reset(latch, expectedCount);
executor.submit(batchEventProcessor);
long start = System.currentTimeMillis();
final RingBuffer<ValueEvent> rb = ringBuffer;
for (long i = 0; i < ITERATIONS; i++) {
long hi = rb.next(BATCH_SIZE);
long lo = hi - (BATCH_SIZE - 1);
for (long l = lo; l <= hi; l++) {
rb.get(l).setValue(i);
}
rb.publish(lo, hi);
}
latch.await();
long opsPerSecond = (BATCH_SIZE * ITERATIONS * 1000L) / (System.currentTimeMillis() - start);
waitForEventProcessorSequence(expectedCount);
batchEventProcessor.halt();
failIfNot(expectedResult, handler.getValue());
return opsPerSecond;
}
private void waitForEventProcessorSequence(long expectedCount) throws InterruptedException {
while (batchEventProcessor.getSequence().get() != expectedCount) {
Thread.sleep(1);
}
}
public static void main(String[] args) throws Exception {
System.out.println("ProducerType.SINGLE, use yield wait strategy");
OneToOneSequencedBatchThroughputTest test1 = new OneToOneSequencedBatchThroughputTest(ProducerType.SINGLE, new YieldingWaitStrategy());
test1.testImplementations();
System.out.println("ProducerType.SINGLE, use sleep wait strategy");
OneToOneSequencedBatchThroughputTest test2 = new OneToOneSequencedBatchThroughputTest(ProducerType.SINGLE, new SleepingWaitStrategy());
test2.testImplementations();
System.out.println("ProducerType.SINGLE, use block wait strategy");
OneToOneSequencedBatchThroughputTest test3 = new OneToOneSequencedBatchThroughputTest(ProducerType.SINGLE, new BlockingWaitStrategy());
test3.testImplementations();
System.out.println("ProducerType.SINGLE, use busy spin wait strategy");
OneToOneSequencedBatchThroughputTest test4 = new OneToOneSequencedBatchThroughputTest(ProducerType.SINGLE, new BusySpinWaitStrategy());
test4.testImplementations();
System.out.println("ProducerType.MULTI, use yield wait strategy");
OneToOneSequencedBatchThroughputTest test5 = new OneToOneSequencedBatchThroughputTest(ProducerType.MULTI, new YieldingWaitStrategy());
test5.testImplementations();
System.out.println("ProducerType.MULTI, use sleep wait strategy");
OneToOneSequencedBatchThroughputTest test6 = new OneToOneSequencedBatchThroughputTest(ProducerType.MULTI, new SleepingWaitStrategy());
test6.testImplementations();
System.out.println("ProducerType.MULTI, use block wait strategy");
OneToOneSequencedBatchThroughputTest test7 = new OneToOneSequencedBatchThroughputTest(ProducerType.MULTI, new BlockingWaitStrategy());
test7.testImplementations();
System.out.println("ProducerType.MULTI, use busy spin wait strategy");
OneToOneSequencedBatchThroughputTest test8 = new OneToOneSequencedBatchThroughputTest(ProducerType.MULTI, new BusySpinWaitStrategy());
test8.testImplementations();
}
}
可以看到在ProducerType.SINGLE, use block wait strategy模式下性能最好。
ProducerType.SINGLE, use yield wait strategy
Starting Disruptor tests
Run 0, Disruptor=77,118,840 ops/sec
Run 1, Disruptor=77,291,698 ops/sec
Run 2, Disruptor=81,994,096 ops/sec
Run 3, Disruptor=82,014,270 ops/sec
Run 4, Disruptor=82,304,526 ops/sec
Run 5, Disruptor=81,967,213 ops/sec
Run 6, Disruptor=82,142,270 ops/sec
ProducerType.SINGLE, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=82,829,454 ops/sec
Run 1, Disruptor=82,856,906 ops/sec
Run 2, Disruptor=82,802,020 ops/sec
Run 3, Disruptor=83,063,377 ops/sec
Run 4, Disruptor=80,205,325 ops/sec
Run 5, Disruptor=80,295,487 ops/sec
Run 6, Disruptor=80,366,471 ops/sec
ProducerType.SINGLE, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=81,433,224 ops/sec
Run 1, Disruptor=82,385,895 ops/sec
Run 2, Disruptor=82,223,318 ops/sec
Run 3, Disruptor=82,590,023 ops/sec
Run 4, Disruptor=82,243,605 ops/sec
Run 5, Disruptor=83,160,083 ops/sec
Run 6, Disruptor=82,966,896 ops/sec
ProducerType.SINGLE, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=80,651,665 ops/sec
Run 1, Disruptor=80,418,174 ops/sec
Run 2, Disruptor=80,906,148 ops/sec
Run 3, Disruptor=81,294,203 ops/sec
Run 4, Disruptor=81,043,844 ops/sec
Run 5, Disruptor=81,347,108 ops/sec
Run 6, Disruptor=81,799,591 ops/sec
ProducerType.MULTI, use yield wait strategy
Starting Disruptor tests
Run 0, Disruptor=52,938,062 ops/sec
Run 1, Disruptor=52,358,762 ops/sec
Run 2, Disruptor=52,770,448 ops/sec
Run 3, Disruptor=53,746,103 ops/sec
Run 4, Disruptor=53,780,789 ops/sec
Run 5, Disruptor=54,112,554 ops/sec
Run 6, Disruptor=53,858,997 ops/sec
ProducerType.MULTI, use sleep wait strategy
Starting Disruptor tests
Run 0, Disruptor=53,078,556 ops/sec
Run 1, Disruptor=52,982,939 ops/sec
Run 2, Disruptor=53,134,962 ops/sec
Run 3, Disruptor=52,356,020 ops/sec
Run 4, Disruptor=53,123,671 ops/sec
Run 5, Disruptor=53,112,385 ops/sec
Run 6, Disruptor=53,129,316 ops/sec
ProducerType.MULTI, use block wait strategy
Starting Disruptor tests
Run 0, Disruptor=37,166,431 ops/sec
Run 1, Disruptor=36,571,094 ops/sec
Run 2, Disruptor=36,549,707 ops/sec
Run 3, Disruptor=36,496,350 ops/sec
Run 4, Disruptor=36,654,204 ops/sec
Run 5, Disruptor=36,986,352 ops/sec
Run 6, Disruptor=36,873,156 ops/sec
ProducerType.MULTI, use busy spin wait strategy
Starting Disruptor tests
Run 0, Disruptor=53,925,798 ops/sec
Run 1, Disruptor=53,676,865 ops/sec
Run 2, Disruptor=53,613,553 ops/sec
Run 3, Disruptor=53,888,020 ops/sec
Run 4, Disruptor=53,789,468 ops/sec
Run 5, Disruptor=53,827,107 ops/sec
Run 6, Disruptor=53,838,699 ops/sec