Rebus 中的“SetMaxParallelism”和“SetNumberOfWorkers”在多大程度上混淆了处理来自两个主题的消息的顺序？

Question

我一直向两个主题发送大量编号消息：

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. 它们由一名工作人员分四个流接收，并在不同的时间内进行处理，从几分之一秒到一分钟不等。

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

请告诉我，Rebus在加工过程中会混合多少？它们的处理顺序可以在主题之间和内部主题之间混合吗？

2. 如果我在下一个配置中接受相同的事件：

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

混合会相同还是不同？

3. 如果我在下一个配置中接受相同的事件：

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

混合会相同还是不同？

4. 如果我在下一个配置中接受相同的事件：

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

混合是相同还是不同？也许工作人员会消费他们主题中的事件？

5. 如果我在下一个配置中接受相同的事件：

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

混合是相同还是不同？也许线程会消耗来自其主题的事件？

如果有针对混合事件处理顺序的某些规则的测试，请提供此测试的链接。

Answer 1

在所有情况下，Rebus 都会按照消息从输入队列传送的顺序简单地处理消息。

如果并行度== 1，那么消息将按照它们在队列中的顺序进行处理(*)。

一旦并行度 > 1，就有可能同时处理多个消息，这使得不可能对顺序做出任何承诺。如果某些消息需要很长时间来处理，它们将比队列中的邻居晚得多。

所以一次性回答您所有的问题：

一个Rebus实例有一个输入队列。它将把消息从输入队列中取出并尽快处理它们。

如果并行度== 1，那么消息将按顺序处理（*）。

如果并行度 > 1，那么就不能期望任何排序。

---

(*)：请注意，如果您假设消息严格按顺序到达，则失败的消息可能会造成严重破坏。当消息失败时（即引发异常），Rebus 会将消息移至其死信队列（默认名称为“error”）。如果将失败的消息移回到 Rebus 的输入队列中，那么它可能会完全无序处理。