将异步消息分组在一起答案

【问题标题】：Grouping asynchronous messages together将异步消息分组在一起
【发布时间】：2021-03-17 23:00:24
【问题描述】：

我希望这个问题的范围不要太广，但这里是：

上下文

我正在尝试在 .NET Core 中创建一个后台服务，该服务从消息队列（在我的情况下为 RabbitMQ）中获取异步和无序消息流，将它们组合在一起并将每组消息作为一条消息发布到再次消息队列。这些消息采用以下格式：

<timestamp> - <group-identifier> - <message-contents>

例如，给定以下消息流：

2020-12-06T10:10:27.556Z - group1 - content1
2020-12-06T10:10:27.557Z - group1 - content2
2020-12-06T10:10:27.558Z - group2 - content1
2020-12-06T10:10:27.558Z - group1 - content3
2020-12-06T10:10:27.570Z - group2 - content2

我希望发布以下消息：

已发布消息 1：

{
  "group": "group1",
  "messages": [
    {
      "timestamp": "2020-12-06T10:10:27.556Z",
      "content": "content1"
    },
    {
      "timestamp": "2020-12-06T10:10:27.557Z",
      "content": "content2"
    },
    {
      "timestamp": "2020-12-06T10:10:27.558Z",
      "content": "content3"
    }
  ]
}

已发布消息 2：

{
  "group": "group2",
  "messages": [
    {
      "timestamp": "2020-12-06T10:10:27.558Z",
      "content": "content1"
    },
    {
      "timestamp": "2020-12-06T10:10:27.570Z",
      "content": "content2"
    }
  ]
}

我想出了什么

我能想到的唯一解决方案是使用缓存（如 Redis）来存储消息，并在一定次数后发布消息当新组的消息进入时，会发生以下情况：

将消息推送到其组标识符下的缓存；
启动一个 1 秒的计时器；
在计时器运行时，当该组的另一条消息进入时，将它们连同组标识符一起推送到缓存中；
定时器结束后，从缓存中取出该组的所有消息并发布它们；

但是，我发现这种方法存在问题。在我的情况下，在发送消息之前等待 1 秒可能就足够了，但远非有效。在理想情况下，所有消息通过只需要 1 或 2 毫秒，但由于网络延迟或代码执行时间过长，可能需要更长时间。但是如果我缩短这段时间，我恐怕会开始丢失消息。

所以现在我正在尝试找到一个解决方案来修复或（至少）缓解这个问题。但到目前为止我还没有运气。这真的是最好的方法，还是我可以做一些不同的事情？

提前致谢！

【问题讨论】：

那么问题是如何避免在发出分组消息之前等待一秒钟？我不认为这是可以解决的，除非你事先知道有多少消息会在接下来的时刻出现，这些消息是相关的并且应该组合在一起。
如果您的队列中有一千条消息，您是否需要处理所有消息以构建新的有效负载？如果您的订阅者倒下并且您有 10 万条消息积压怎么办？

标签： c# asynchronous .net-core rabbitmq message-queue

【解决方案1】：

不是 100% 确定我理解了这个问题。您可以在应用程序中创建线程安全的单例字典。每当您通过消息代理（在您的情况下为 RabbitMQ）收到消息时 -> 尝试将其添加到字典中，如果它已经存在 -> 通过使用相同的键将其添加到集合中来修改值。当您添加/修改值时，将逻辑添加到您的字典（通过继承 ConccurentDictionary 类（它是线程安全的）来创建您的类型） -> 检查数组中有多少项具有该键。然后设置一些规则，例如，一旦数组中有 10 个项目 - 将消息发布到下一个队列并从字典中删除该键值对

【讨论】：