Use Disruptor to implement our lightweight queue model

[CalvinKirs]

• Why do you submit this issue?

• Question or discussion

• Feature or performance improvement

Requirement or improvement

Hi, guys, in #6744, the @wu-sheng teacher suggested that we can use Disruptor to implement our lightweight queue model, the project is widely used and the performance looks great.

Everyone is very welcome to discuss
If everyone agrees on ， I would like to finish this work. ( I will provide JMH-Test after the implementation)

disruptor
GitHub: https://github.com/LMAX-Exchange/disruptor

License: Apache-2.0

[wu-sheng]

@CalvinKirs With months passed, any update? If not, it is better to close this issue, or we at least should convert this to discussion.

[CalvinKirs]

@CalvinKirs With months passed, any update? If not, it is better to close this issue, or we at least should convert this to discussion.

I'm sorry I've been so busy lately, I'll close it for now and I'll sync here when I've made progress.

[CalvinKirs]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

[wu-sheng]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

Disruptor has better performance for sure. We have tested years ago. The challenge is thread-model. We are not using one thread per logic queue, because we created one logic queue per metrics, which could be changed and keep increasing.

[CalvinKirs]

I did a simple test comparison and it looks like Disruptor does serve our purpose, I will provide the full JMH Test shortly.

Disruptor has better performance for sure. We have tested years ago. The challenge is thread-model. We are not using one thread per logic queue, because we created one logic queue per metrics, which could be changed and keep increasing.

Yes, this is a clever design by Channel,

For Disruptor, at present, I can only think of providing a more flexible configuration to achieve an ideal effect.

[CalvinKirs]

testSingleThreadDataCarrier:
Single-threaded message delivery
2 threads consumption
Data volume=100*dataSize
channelSize=32 bufferSize=4

testSingleThreadMiniDataCarrier:
Single-threaded message delivery
2 threads consumption
Data volume=100*dataSize
channelSize=4 bufferSize=32

testSingleThreadDisruptor
Single-threaded message delivery
Single producer
2 threads consumption
Data volume=100*dataSize
ringBufferSize=128

testMultiThreadDataCarrier
2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
channelSize=32 bufferSize=4
testMultiThreadMiniChannelDataCarrier

2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
channelSize=4 bufferSize=32

testMultiThreadDisruptor
Multi-producer mode (only one thread is used)
2 threads to deliver messages
2 threads consumption
Data volume=100*2*dataSize
ringBufferSize=128

Benchmark                                                                                             (dataSize)   Mode  Cnt         Score   Error   Units
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                       1  thrpt    2        21.587           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                        1  thrpt    2         0.026          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                   1  thrpt    2      1331.032            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space                               1  thrpt    2        27.020          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space.norm                          1  thrpt    2   1378235.838            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space                           1  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space.norm                      1  thrpt    2      8275.731            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                             1  thrpt    2         4.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.time                                              1  thrpt    2         9.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                      10  thrpt    2         2.616           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                       10  thrpt    2         0.003          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                  10  thrpt    2      1363.219            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space                              10  thrpt    2         6.106          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Eden_Space.norm                         10  thrpt    2   2591192.615            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space                          10  thrpt    2         0.170          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.churn.PS_Survivor_Space.norm                     10  thrpt    2     72070.769            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                            10  thrpt    2         1.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.time                                             10  thrpt    2         3.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                     100  thrpt    2         0.273           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                      100  thrpt    2        ≈ 10⁻³          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                 100  thrpt    2      1632.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                           100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier                                                    1000  thrpt    2         0.027           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate                                     1000  thrpt    2        ≈ 10⁻⁴          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.alloc.rate.norm                                1000  thrpt    2      4300.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDataCarrier:·gc.count                                          1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                         1  thrpt    2      8933.128           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                          1  thrpt    2        11.479          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                     1  thrpt    2      1416.457            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                                 1  thrpt    2     14422.861          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                            1  thrpt    2   1778182.066            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                             1  thrpt    2         1.161          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                        1  thrpt    2       143.073            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                               1  thrpt    2      1788.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                                1  thrpt    2       604.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                        10  thrpt    2      2444.081           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                         10  thrpt    2         4.281          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                    10  thrpt    2      1928.139            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                                10  thrpt    2      3154.202          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                           10  thrpt    2   1421635.320            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                            10  thrpt    2         0.256          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                       10  thrpt    2       115.265            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                              10  thrpt    2       391.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                               10  thrpt    2       171.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                       100  thrpt    2       320.904           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                        100  thrpt    2         2.078          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                   100  thrpt    2      7115.277            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                               100  thrpt    2       378.837          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                          100  thrpt    2   1300370.489            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                           100  thrpt    2         0.024          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                      100  thrpt    2        81.640            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                             100  thrpt    2        47.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                              100  thrpt    2        21.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor                                                      1000  thrpt    2        32.394           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate                                       1000  thrpt    2         1.741          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.alloc.rate.norm                                  1000  thrpt    2     59387.583            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space                              1000  thrpt    2       387.073          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Eden_Space.norm                         1000  thrpt    2  13159086.907            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space                          1000  thrpt    2         0.031          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                     1000  thrpt    2      1061.780            B/op
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.count                                            1000  thrpt    2        48.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadDisruptor:·gc.time                                             1000  thrpt    2        23.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                            1  thrpt    2        19.972           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                             1  thrpt    2         0.024          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                        1  thrpt    2      1331.280            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space                    1  thrpt    2        25.725          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space.norm               1  thrpt    2   1419509.760            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space                1  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space.norm           1  thrpt    2      8937.580            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                  1  thrpt    2         4.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.time                                   1  thrpt    2        11.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                           10  thrpt    2         2.499           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                            10  thrpt    2         0.003          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                       10  thrpt    2      1358.763            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space                   10  thrpt    2         5.905          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Eden_Space.norm              10  thrpt    2   2591192.615            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space               10  thrpt    2         0.162          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.churn.PS_Survivor_Space.norm          10  thrpt    2     70975.385            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                 10  thrpt    2         1.000          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.time                                  10  thrpt    2         3.000              ms
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                          100  thrpt    2         0.256           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                           100  thrpt    2        ≈ 10⁻³          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                      100  thrpt    2      1557.333            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                                100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier                                         1000  thrpt    2         0.028           ops/s
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate                          1000  thrpt    2        ≈ 10⁻⁴          MB/sec
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.alloc.rate.norm                     1000  thrpt    2      2092.000            B/op
DisruptorDataCarrierBenchmark.testMultiThreadMiniChannelDataCarrier:·gc.count                               1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                      1  thrpt    2        43.171           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                       1  thrpt    2         0.220          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                  1  thrpt    2      5601.517            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                            1  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                     10  thrpt    2         5.243           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                      10  thrpt    2         0.268          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                 10  thrpt    2     56349.585            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                           10  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                    100  thrpt    2         0.530           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                     100  thrpt    2         0.307          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                                100  thrpt    2    635533.333            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                          100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier                                                   1000  thrpt    2         0.053           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate                                    1000  thrpt    2         0.316          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.alloc.rate.norm                               1000  thrpt    2   6426032.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDataCarrier:·gc.count                                         1000  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                        1  thrpt    2     23499.217           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                         1  thrpt    2       120.315          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                    1  thrpt    2      5640.267            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                                1  thrpt    2       122.342          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                           1  thrpt    2      5735.384            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                            1  thrpt    2         0.015          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                       1  thrpt    2         0.694            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                              1  thrpt    2        16.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                               1  thrpt    2        11.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                       10  thrpt    2      5626.929           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                        10  thrpt    2       287.682          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                   10  thrpt    2     56325.927            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                               10  thrpt    2       289.150          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                          10  thrpt    2     56627.705            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                           10  thrpt    2         0.012          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                      10  thrpt    2         2.343            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                             10  thrpt    2        36.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                              10  thrpt    2        16.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                      100  thrpt    2       750.957           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                       100  thrpt    2       383.967          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                  100  thrpt    2    563272.376            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                              100  thrpt    2       386.852          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                         100  thrpt    2    567547.589            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                          100  thrpt    2         0.019          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                     100  thrpt    2        28.311            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                            100  thrpt    2        48.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                             100  thrpt    2        22.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor                                                     1000  thrpt    2        82.218           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate                                      1000  thrpt    2       473.891          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.alloc.rate.norm                                 1000  thrpt    2   6348470.312            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space                             1000  thrpt    2       475.561          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Eden_Space.norm                        1000  thrpt    2   6371246.361            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space                         1000  thrpt    2         0.040          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.churn.PS_Survivor_Space.norm                    1000  thrpt    2       537.732            B/op
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.count                                           1000  thrpt    2        59.000          counts
DisruptorDataCarrierBenchmark.testSingleThreadDisruptor:·gc.time                                            1000  thrpt    2        27.000              ms
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                  1  thrpt    2        43.311           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                   1  thrpt    2         0.220          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                              1  thrpt    2      5601.513            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                        1  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                 10  thrpt    2         5.194           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                  10  thrpt    2         0.266          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                             10  thrpt    2     56316.615            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                       10  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                                100  thrpt    2         0.527           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                 100  thrpt    2         0.306          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                            100  thrpt    2    635616.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                      100  thrpt    2           ≈ 0          counts
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier                                               1000  thrpt    2         0.054           ops/s
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate                                1000  thrpt    2         0.325          MB/sec
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.alloc.rate.norm                           1000  thrpt    2   6426080.000            B/op
DisruptorDataCarrierBenchmark.testSingleThreadMiniDataCarrier:·gc.count                                     1000  thrpt    2           ≈ 0          counts

[wu-sheng]

Surprising to see this update. Could you be clear about which cases are you using?

[CalvinKirs]

Sorry for the late reply.
I have some questions, our current consumption is to pull the list data in the queue, and the Disruptor is to process the data one by one.
For some operations designed to DB, do we need to take some measures to do batch processing and consumption?

[wu-sheng]

Yes, batch-mode is important. But I remember disruptor has end-of-batch concept to cover this case.

Also, you should notice, in the latest OAP side DataCarrier(right now, OAP and Agent DataCarrier are different) has a concept of IConsumer#nothingToConsume event, which I think missed in Disruptor. It is important to build batch(not to database) as it is based on both duration and size.

[CalvinKirs]

Surprising to see this update. Could you be clear about which cases are you using?

Thanks for your reply, am I missing something?

 public static AtomicInteger oldQueueConsumerIndex = new AtomicInteger(0);
    public static AtomicInteger disruptorConsumerIndex = new AtomicInteger(0);
    DataCarrier<String> dataCarrier = new DataCarrier<>(32, 4);
    DataCarrier<String> miniChannelDataCarrier = new DataCarrier<>(4, 32);
    DataCarrierDisruptor disruptor = new DataCarrierDisruptor("Message", TestWorkHandler.class, ProducerType.SINGLE);
    DataCarrierDisruptor multiProducerDisruptor = new DataCarrierDisruptor("Message", TestWorkHandler.class, ProducerType.MULTI);


    @Param(value = {"1", "10", "100", "1000"})
    private int dataSize;


    @Setup
    public void setUp() {
        dataCarrier.consume(TestOldConsumer.class, 2);
        miniChannelDataCarrier.consume(TestOldConsumer.class, 2);
    }

    @Benchmark
    public void testSingleThreadDataCarrier() {
        for (int i = 0; i < 100 * dataSize; i++) {
            dataCarrier.produce("Message:" + i);
        }
        while (oldQueueConsumerIndex.get() < 100 * dataSize) {
            Thread.yield();
        }
        oldQueueConsumerIndex.set(0);
    }

public class TestOldConsumer implements IConsumer<String> {


    @Override
    public void init(Properties properties) {

    }

    @Override
    public void consume(List<String> data) {
        if (CollectionUtils.isEmpty(data)) {
            return;
        }
        data.forEach(d -> {
            DisruptorDataCarrierBenchmark.oldQueueConsumerIndex.getAndIncrement();
        });
    }
}