azure - 使用 eventhub 时，为什么我会在 Azure 上收到此错误？

Question

我最近开始使用 Azure，这是一次压倒性的体验。我开始尝试 eventhubs我基本上遵循 official tutorials关于如何使用 Nodejs 从 eventhub 发送和接收消息。

一切都很完美，所以我构建了一个小型 Web 应用程序(静态前端应用程序)，并将其与节点后端连接，在后端与 eventhub 进行通信。所以基本上我的应用程序是这样构建的:

frontend <----> node server <-----> eventhubs

正如你所看到的，它非常简单。节点服务器从 eventhub 获取数据并将其转发到前端，其中显示值。这是一次很酷的体验，我很喜欢 MS Azure，直到出现此错误:

 azure.eventhub.common.EventHubError: ErrorCodes.ResourceLimitExceeded: Exceeded the maximum number of allowed receivers per partition in a consumer group which is 5. List of connected receivers - nil, nil, nil, nil, nil.

这个错误确实令人困惑。我使用默认的消费者组并且仅使用一个应用程序。我从未尝试从其他应用程序访问这个消费者组。它说限制是 5，我只使用一个应用程序，所以应该没问题，还是我遗漏了一些东西？我不会检查这里发生了什么。

我浪费了太多时间在谷歌上搜索和研究这个问题，但我没有明白。最后，我想也许每次我在 azure 上部署应用程序(我的前端和节点服务器)时，这都会被算作一个消费者，并且由于我部署应用程序超过 5 次，所以就会出现此错误。我是对的还是这是胡说八道？

编辑

我使用 websockets 作为我的应用程序(前端)和节点服务器(后端)之间的通信协议(protocol)。节点服务器正在使用默认的消费者组(我没有更改任何内容)，我只是按照this official example from Microsoft 。我基本上使用的是 MS 文档中的代码，这就是为什么我没有从节点服务器发布任何代码片段，并且由于错误发生在后端而不是前端，所以如果我发布任何前端代码将没有帮助。

总而言之，我使用 websocket 来连接前端和后端。它完美地工作了一两天，然后这个错误开始发生。有时我会打开多个客户端(例如来自浏览器的客户端和来自智能手机的客户端)。

我想我不太理解这个消费群体的概念。就像每个客户都是消费者吗？因此，如果我在浏览器的 5 个不同选项卡中打开我的应用程序(同一个应用程序)，那么我是否有 5 个消费者？

<小时/>

我不太理解下面的答案以及“池客户端”的含义，因此，我将尝试在此处发布代码示例来向您展示我正在尝试做什么。

代码片段

这是我在服务器端使用的函数，用于与 eventhub 通信并接收/使用消息

async function receiveEventhubMessage(socket, eventHubName, connectionString) {

 
  const consumerClient = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName);


  const subscription = consumerClient.subscribe({
      processEvents: async (events, context) => {

        for (const event of events) {
          console.log("[    consumer    ] Message received : " + event.body);

            io.emit('msg-received', event.body);
        }
      },

      processError: async (err, context) => {
        console.log(`Error : ${err}`);
      }
    }
  );

如果您注意到，我将 eventhub 和连接字符串作为参数提供，以便能够更改它。现在在前端，我有一个包含多个主题的列表，每个主题都有自己的 eventhubname，但它们具有相同的 eventhub 命名空间。

这是我拥有的两个 eventhubname 的示例:

{
"EventHubName": "eh-test-command"
"EventHubName": "eh-test-telemetry"
}

如果用户选择发送命令(从前端，我只有一个按钮列表，用户可以单击这些按钮通过 Websockets 触发事件)，那么 CommandEventHubName 将从节点服务器的前端。服务器将接收该 eventhubname 并在我上面发布的函数中切换 ConsumerClient。

这是我调用它的代码:

// io is a socket.io object
io.on('connection', socket => {
   socket.on('onUserChoice', choice => {
   // choice is an object sent from the frontend based on what the user choosed. e.g if the user choosed command then choice = {"EventhubName": "eh-test-command", "payload": "whatever"}

   receiveEventhubMessage(socket, choice.EventHubName, choice.EventHubNameSpace)
              .catch(err => console.log(`[    consumerClient    ] Error while receiving eventhub messages: ${err}`));
      }
}

我正在构建的应用程序将来将扩展到汽车领域的真实用例，这就是为什么这对我来说很重要。因此，我试图弄清楚如何在 eventhub 之间进行切换，而无需每次 eventhubname 更改时创建新的 ConsumerClient？

我必须说我不理解“池客户端”的示例。我正在寻求更多的阐述，或者最好是一个最小的例子，只是为了让我上路。

Answer 1

根据问题中的对话，其根本原因似乎是您的后端正在为来自前端的每个请求创建一个新的 EventHubConsumerClient。由于每个客户端都会打开与服务的专用连接，因此如果您使用同一使用者组对同一事件中心实例发出超过 5 个请求，则会超出配额。

要解决此问题，您需要考虑池化 EventHubConsumerClient 实例，以便每个事件中心实例都有一个实例。您可以通过调用 subscribe 安全地使用池客户端来处理前端请求。这将允许您在多个前端请求之间共享连接。

关键思想是您的 consumerClient 不是为每个请求创建的，而是在请求之间共享一个实例。使用您的代码片段来说明最简单的方法，您最终会将客户端创建提升到要接收的函数之外。它可能看起来像:

const consumerClient = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName);

async function receiveEventhubMessage(socket, eventHubName, connectionString) {
  const subscription = consumerClient.subscribe({
      processEvents: async (events, context) => {

        for (const event of events) {
          console.log("[    consumer    ] Message received : " + event.body);
          io.emit('msg-received', event.body);
        }
      },

      processError: async (err, context) => {
        console.log(`Error : ${err}`);
      }
    }
  );

也就是说，根据应用程序的架构，上述内容可能不足以满足您的环境。如果托管 receiveEventHubMessage 的内容是为每个请求动态创建的，则不会发生任何变化。在这种情况下，您需要考虑使用单例或依赖注入(inject)之类的方法来帮助延长生命周期。

如果您最终在扩展以满足您的请求时遇到问题，您可以考虑增加每个事件中心的客户端数量和/或将请求分散到不同的消费者组。