Rebus中SetMaxParallelism与SetNumberOfWorkers对消息处理顺序的影响

问题1 #

发送代码如下：

for (var i=0; i<500; i++)
{
    await bus.Publish(new TestEvent1 { Order = i });
}
for (var i=0; i<500; i++)
{
    await bus.Publish(new TestEvent2 { Order = i });
}

消息被1个Worker以4个流接收，处理时长从数秒到一分钟不等，配置如下：

services.AddRebus((configurer, serviceProvider) => configurer
    ...
    .Options(o =>
    {
        o.SetMaxParallelism(4);
        o.SetNumberOfWorkers(1);
    });

请问Rebus处理时会将消息顺序打乱到什么程度？主题之间及主题内部的处理顺序是否都会混乱？

回答 #

• 主题内部顺序一定会混乱：单个Worker开启4个并行处理流，消息按批次拉取后并行处理，由于每个消息处理时长差异大，先拉取的消息可能因处理慢，比后拉取的消息完成得晚，比如TestEvent1的Order=0可能在Order=10之后处理完成。
• 主题之间顺序也会混乱：两个主题的消息会被混合拉取并行处理，TestEvent1和TestEvent2的消息会交叉完成，比如TestEvent2的Order=0可能在TestEvent1的Order=50之前处理完成。
• 整体混乱程度高，所有消息的处理完成顺序几乎完全取决于各自的处理时长，和发送顺序关联性极低。

问题2 #

若采用以下配置接收相同事件：

services.AddRebus((configurer, serviceProvider) => configurer
    ...
    .Options(o =>
    {
        o.SetMaxParallelism(1);
        o.SetNumberOfWorkers(4);
    });

消息处理顺序的混乱情况会与场景1相同还是不同？

回答 #

和场景1不同，但整体仍会混乱，只是混乱逻辑有差异：

• 每个Worker线程都是单并行处理，即每个Worker会按顺序处理自己拉取到的消息，但4个Worker会各自独立拉取消息。
• 主题内部：单个Worker拉取到的TestEvent1消息会按顺序处理，但不同Worker可能拉取不同批次的TestEvent1，比如Worker1处理Order=0、1，Worker2处理Order=5、6，那么Worker2的Order=5可能在Worker1的Order=1之前完成，所以主题内部顺序还是乱的。
• 主题之间：不同Worker可能同时处理TestEvent1和TestEvent2的消息，跨主题的顺序也会混乱。
• 对比场景1：场景1是单线程内4个并行流，场景2是4个单并行线程，最终处理完成的混乱程度类似，但底层调度逻辑不同，外部表现都是全量顺序混乱。

问题3 #

若采用以下配置接收相同事件：

services.AddRebus((configurer, serviceProvider) => configurer
    ...
    .Options(o =>
    {
        o.SetMaxParallelism(2);
        o.SetNumberOfWorkers(2);
    });

消息处理顺序的混乱情况会与之前场景相同还是不同？

回答 #

和场景1、2的混乱程度和逻辑类似，本质都是总并行度为4（2*2），只是拆分方式不同：

• 总共有2个Worker线程，每个线程可并行处理2个消息，整体并行处理能力还是4个。
• 不管是单线程多并行还是多线程单并行，只要总并行度一致，最终的处理顺序混乱表现几乎无区别——主题内部、主题之间的顺序都会混乱，完成顺序完全由消息处理时长决定。

问题4 #

若采用以下配置接收相同事件：

services.AddRebus((configurer, serviceProvider) => configurer
    ...
    .Options(o =>
    {
        o.SetMaxParallelism(1);
        o.SetNumberOfWorkers(2);
    });

消息处理顺序的混乱情况会与之前场景相同还是不同？是否每个Worker会单独消费对应主题的事件？

回答 #

• 混乱情况和场景2类似（总并行度2），但因总并行度比场景1、3低，整体混乱程度稍低，但主题内部和跨主题的顺序依然会混乱。
• 不会每个Worker单独消费对应主题：Rebus的Worker从同一个输入队列拉取消息，不会按主题分配Worker。两个Worker都会随机拉取TestEvent1和TestEvent2的消息，各自按顺序处理自己拉到的消息，但不同Worker的消息处理完成顺序依然会交叉。

问题5 #

若采用以下配置接收相同事件：

services.AddRebus((configurer, serviceProvider) => configurer
    ...
    .Options(o =>
    {
        o.SetMaxParallelism(2);
        o.SetNumberOfWorkers(1);
    });

消息处理顺序的混乱情况会与之前场景相同还是不同？是否每个线程会单独消费对应主题的事件？

回答 #

• 混乱情况和问题4类似（总并行度2），主题内部和跨主题顺序都会混乱，只是底层是单线程内2个并行流处理，和问题4的双线程单并行逻辑不同，但外部表现的混乱程度一致。
• 不会每个线程单独消费对应主题：单Worker线程内的并行流从同一个队列拉取所有主题的消息，并行处理时不会区分主题，TestEvent1和TestEvent2的消息会被混合分配到不同的并行流中处理。

内容的提问来源于stack exchange，提问作者Mentor