包含来自消费者的 ACK 的较小任务的任务

Question

我正在研究一些 POC project 并尝试解决以下问题。

我有一个 Publisher 正在向队列发送消息：

bus.PublishAsync<IBaseScenario>(new TestScenario())
            .ContinueWith(task =>
            {
                if (task.IsCompleted && !task.IsFaulted)
                    Console.WriteLine("TestScenario queued with success.");
                else
                    Console.WriteLine(task.Exception.Message);
            });

还有一些 Consumers 正在消费消息：

bus.SubscribeAsync<IBaseScenario>("test_1_consumer",
            message => Task.Factory.StartNew(() =>
            {
                var testScenario = message as TestScenario;
                var anotherTestScenario = message as AnotherTestScenario;

                ResolveScenario(testScenario);
                ResolveScenario(anotherTestScenario);

            }).ContinueWith(task =>
            {
                if (task.IsCompleted && !task.IsFaulted)
                    Console.WriteLine("Task ended up with success.");
                else
                    Console.WriteLine(task.Exception.Message);
            }));

此时一切都按需要进行，但这是我想要实现的目标。

我的Message 是某种包含步骤的场景，每个场景都被发送到队列，然后由消费者维护。

1. 每次在消费者网站上完成每个步骤时，我都想从消费者那里获得某种ACK 信息发送给发布者（例如，如果它最终成功与否。

2. 我还想获得有关哪个消费者收到消息的信息。

每条消息（场景）都应该被视为原子操作，因此不应该在不同的消费者上执行步骤，如果某个步骤没有成功结束，那么整个场景应该被视为失败。

是否可以使用以下架构来解决这两个要求，还是我需要使用更多的东西？

Answer 1

最简单的方法是使用这里描述的 EasyNetQ 的请求响应模型https://github.com/EasyNetQ/EasyNetQ/wiki/Request-Response

在响应中，您可以输入处理消息的消费者的身份和场景的最终状态。如果一个场景在一条消息中发送，并且该场景包含所有必要的步骤，那么所有步骤都将由单个消费者处理。

也就是说，消息重复总是一个问题，因为要么发送消息两次，要么在消费者失败后重新排队消息。如果一个场景永远不会被多次处理是至关重要的，那么您将需要实现消息重复数据删除或使每个场景具有幂等性。这是使用 RabbitMQ 时的普遍事实。