同步消息队列答案

【问题标题】：Synchronized message queue同步消息队列
【发布时间】：2017-03-22 01:21:36
【问题描述】：

我有一个系统需要完成一组操作，每个操作都有一个类型和一个实体。每个实体的操作可以独立完成，但在实体内，类型的顺序确实很重要。一个类型的操作只有在前一个类型的所有操作都完成后才能开始。操作通常会从数据库中获取一些数据，对其进行处理，然后将结果再次存储到数据库中。

假设操作量需要并行使用多台机器，我如何才能实现这种效果（最好在 .NET 世界中）？

附：现在我正在使用服务总线，每条消息都是一个操作实体类型的组合。这根本不符合我的需求，因为订单没有被强制执行，并且不清楚与消息相关的操作何时真正完成。

编辑：我需要找到一种合乎逻辑的方式来以正确的顺序处理这些操作，以保持一切（可以）并行。

示例：我需要为 2 个实体（E1、E2）执行 3 种类型的操作（T1-T3）。可以有许多具有相同类型和实体的操作。

-- Entity 1
[T1, E1][T1, E1][T1, E1][T1, E1] - 4 operations Type 1, Entity 1
[T2, E1][T2, E1] - 2 ops, Type 2, Entity 1
[T3, E1] - 1 operation, Type 3, Entity 1
-- Entity 2
[T2, E2][T2, E2]
[T3, E2]

通常我可以并行处理实体 1 和实体 2。我也可以并行处理每个特定的 [Tx, Ex]（例如所有 [T2, E1]）。我需要保持类型的顺序，T1->T2->T3。 T2 操作需要等到实体内的所有 T1 操作完成。

【问题讨论】：

您是否需要对这些消息同步提供一些响应，还是一劳永逸？
消息携带有关需要完成的工作的信息，结果存储在数据库中。
因此，当消息到达时，您可以查找（在内存或数据库中）给定实体是否处于正确状态以应用操作。如果是 - 申请。如果不是 - 在内存中缓存操作。当另一条消息到达时，执行相同的操作，但如果该实体有更早到达但尚未应用的挂起操作，则在内存中查找。如果是 - 为它们执行相同的序列（如果现在可以应用 - 应用，否则等到下一次）。
您建议通过数据库进行同步。在机器重新启动/程序崩溃/服务重新启动后，将消息存储在 ram 中会出现问题，因为这会导致消息丢失并且无法重新启动它们。
是的，但是无论如何您都需要使用一些持久性存储，因为如果消息到达并且服务在您将其应用到数据库之前崩溃了怎么办 - 它也会丢失。除非您使用一些消息代理（如 RabbitMQ），它允许您在实际处理消息时通知队列管理器（兔子），并且可以将其从存储中删除。

标签： c# .net message-queue

【解决方案1】：

也许您应该为每种类型的操作（消息）使用单独的队列。您按顺序处理此队列，并且仅当前一个队列为空时。例如。有 3 个队列 A、B 和 C。消费者检查 A 队列中是否有任何消息。如果是 - 消费者得到它并处理。如果不是 - 它检查 B 队列。等等。但是来自 A 队列的最后一条消息仍然可能存在问题 - 当来自 B 队列的消息开始时，它可能仍在处理。

类似的解决方案 - 使用消息优先级。但是最后一条消息的问题仍然存在。

【讨论】：

这将需要额外的同步，尤其是当您引入并行性并且最后一条消息变为最后 n 条消息时。

【解决方案2】：

您如何在您的实体中保留一个状态，该状态会导致允许执行哪些类型的操作。

您将有一些工作人员将操作分配到不同的线程上，并在不再需要执行某种类型的操作时更改状态。

我认为可以检查启动的线程是否已完成，但您甚至可能还想为操作添加状态（例如 Queued、InProgress、Done）。

你可以只查询一个实体是否还有某种类型的操作不是“完成”，当它们都完成时，更新实体的状态，允许状态为 Queued 的下一个操作类型出列并进行处理。

注意

这仅在不变量认为不可能从一种状态转到前一种状态时才有效，这意味着这仅在实体不可能在之后处理 T1 操作时才有效它已经完成了T2操作。在实践中，这意味着不应允许实体在开始处理后获得分配的新操作。

【讨论】：

我了解您建议将其分布在一台机器上的线程上。不过，这不会很好地扩展。很抱歉没有在问题中指定这个“要求”。
您能否详细说明机器是如何分配任务的？
现在？ Azure 服务总线。对于这种情况，最好的方法是什么 - 我不知道。