如何从 Azure 存储队列触发容器应用程序？

Question

我对在 NodeJS 中开发和使用容器有点陌生。

我正在用 NodeJS 开发一个小应用程序，它监听 Azure 存储队列。现在我希望该应用程序由队列中的新消息触发，但我不知道该怎么做。我打算使用 Azure 容器应用程序。

我在 Microsoft SDK 中看到了可用于从队列中读取消息的方法，但它是命令式的。也许有计时器的东西。当队列中有新消息时，是否可以在容器应用程序中执行功能？

有人知道吗？

Answer 1

为此，您可以采用两种通用方法：

1. 使用事件源 SDK（如您所述）。这与特定 SDK 使其成为的命令式（或声明式）一样。在这种情况下，它是 Azure 存储队列 SDK，您需要类似的东西

   const sleep = t => new Promise(resolve => setTimeout(resolve, t));
   
   while (true) {
       const receivedMessages = await queueClient.receiveMessages();
   
       if (receivedMessages.receivedMessageItems.length > 0) {
           for (const msg of receivedMessages.receivedMessageItems) {
               // call your logic to handle and delete the msg                
               // await queueClient.deleteMessage(message.messageId, message.popReceipt);
           }
       } else {
           // queue is empty, check in 5 seconds
           await sleep(5000)
       }
   }
   

   2. 使用一些其他框架为您抽象化，并在消息上调用您的方法。例如，dapr 或 Azure Functions。
   • 对于达普，创建一个看起来像 https://docs.dapr.io/reference/components-reference/supported-bindings/storagequeues/ 的 dapr 组件（请注意，如果您在 Azure 容器应用程序上部署，则必须使用 Azure 的 ARM（或二头肌）或使用 az cli 来定义该组件。This example creates a blob component，但是它应该与队列组件非常相似。所以你将拥有一个看起来像这样的 dapr 组件：

   # dapr-queue-component.yaml
   apiVersion: dapr.io/v1alpha1
   kind: Component
   metadata:
     name: ricardos-queue
   spec:
     type: bindings.azure.storagequeues
     version: v1
     metadata:
     - name: accountName
       value: "yourAccountName"
     # you can use managed identity in Azure to not have to set this here
     # the doc linked above shows how to do that.
     # - name: accountKey
     #  value: "***********"
     - name: queueName
       value: "myqueue"
   

   然后创建它：

   ➜ az containerapp env dapr-component set 
       --name $CONTAINERAPPS_ENVIRONMENT --resource-group $RESOURCE_GROUP 
       --dapr-component-name ricardos-queue 
       --yaml dapr-queue-component.yaml
   

   最后您的代码将如下所示 (see dapr docs for more details)：

   //dependencies 
   import { DaprServer, CommunicationProtocolEnum } from '@dapr/dapr'; 
   
   //code
   const daprHost = "127.0.0.1"; 
   const serverHost = "127.0.0.1";
   const serverPort = "6002"; 
   const daprPort = "3602"; 
   
   start().catch((e) => {
       console.error(e);
       process.exit(1);
   });
   
   async function start() {
       const server = new DaprServer(serverHost, serverPort, daprHost, daprPort, CommunicationProtocolEnum.HTTP);
       await server.binding.receive('ricardos-queue', async (msg) => {
           // handle msg
       });
       await server.startServer();
   }
   

   • 对于 Azure 函数：您可以安装 Azure Functions cli 然后执行

   ➜ func init . --javascript
   ➜ func new --template "Azure Queue Storage trigger" --name my-function-name
   ➜ cat my-function-name/function.json
   {
     "bindings": [
       {
         "name": "myQueueItem",
         "type": "queueTrigger",
         "direction": "in",
         "queueName": "js-queue-items", # change to your queue name
         "connection": "CONNECTION_STRING" # env var with a connection string in it. or value in local.settings.json
       }
     ]
   }
   
   ➜ cat my-function-name/index.js
   
   module.exports = async function (context, msg) {
       # handle msg
   };
   

   在 Azure 容器应用程序上部署时，您可以将缩放规则设置为类似这样的内容（see docs here 有关定义缩放规则的更多详细信息，以便如果队列中没有消息，您的应用程序将缩放到 0）

       "scale": {
           "maxReplicas": 10,
           "minReplicas": 0,
           "rules": [
             {
               "name": "my-queue-scale-rule",
               "custom": {
                 "type": "azure-queue",
                 "metadata": {
                   "queueName": "my-queue",
                   "queueLength": "50" // each instance can handle up to 50 messages with a max of 10 instances
                 },
                 "auth": [
                   {
                     "secretRef": "my-connection-string",
                     "triggerParameter": "connection"
                   }
                 ]
               }
             }
           ]
         }
   

   无论您使用的是什么 SDK 或框架（Azure 存储 SDK、dapr、Azure Functions 等），您始终可以使用这样的缩放规则来根据负载向上和向下缩放队列处理器