如何让我的消费者使用 Azure sdk for Golang v0.3.1 监听 Azure 服务总线上的消息？

Posted 2023-02-19

【中文标题】如何让我的消费者使用 Azure sdk for Golang v0.3.1 监听 Azure 服务总线上的消息？

【英文标题】：How do I keep my consumer listening the messages on Azure Sevice Bus using Azure sdk for Golang v0.3.1?

【发布时间】：2022-01-13 17:58:22

【问题描述】：

我一直在使用模块 azure-sdk-for-go/sdk/messaging/azservicebus v0.3.1 将我的使用者与 Azure 服务总线连接，但实现的代码仅接收固定数量的消息，然后停止应用程序我想让消费者一直在听队列。按照我的代码：

    client, err := azservicebus.NewClientFromConnectionString("Connection String", nil)
    
    if err != nil 
        log.Fatalf("Failed to create Service Bus Client: %s", err.Error())
    

    receiver, err := client.NewReceiverForQueue("queue", nil)

    if err != nil 
        log.Fatalf("Failed to create Consumer: %s", err.Error())
    

    messages, err := receiver.ReceiveMessages(context.TODO(), 10, nil)

    if err != nil 
        log.Fatalf("Failed to receive Messages: %s", err.Error())
    

    for _, message := range messages 

        body, err := message.Body()

        if err != nil 
            log.Fatalf("Failed to parse message body: %s", err.Error())
        

        fmt.Println("Message --->", string(body))

        err = receiver.CompleteMessage(context.TODO(), message)

        if err != nil 
            log.Fatalf("Failed to complete message: %s", err.Error())
        

        fmt.Printf("Received and completed message\n")

    

【问题讨论】：

那么，您不必多次致电messages, err := receiver.ReceiveMessages(context.TODO(), 10, nil) 吗？

我不想将 for 放在代码中。我真的很想把责任留给 sdk 模块。

很高兴您的问题已得到解决。您可以接受并投票赞成答案，以便对可能面临相同问题的其他社区成员有所帮助。

【参考方案1】：

您拥有 99% 的代码 - 正如您所指出的，ReceiveMessages 只返回您请求的数量。

所以你只想将你的逻辑包装在一个循环中，这样你就可以连续运行它：

for 
    messages, err := receiver.ReceiveMessages(context.TODO(), 10, nil)
    if err != nil 
        log.Fatalf("Failed to receive Messages: %s", err.Error())
    

    for _, message := range messages 
        body, err := message.Body()

        if err != nil 
            log.Fatalf("Failed to parse message body: %s", err.Error())
        

        fmt.Println("Message --->", string(body))

        err = receiver.CompleteMessage(context.TODO(), message)

        if err != nil 
            log.Fatalf("Failed to complete message: %s", err.Error())
        
    

顺便说一句，0.3.2 刚刚发布，其中包含一些您会感兴趣的修复，因为它们确实会影响 ReceiveMessages。

【讨论】：

另外，@danielfarrell 有正确的答案，但我显然没有足够的声誉来发表评论。

我不想将for 放在代码中。我真的很想把责任留给 sdk 模块。

@RafaelReis，API 做流量控制没有意义。轮询队列是一个不确定的操作。你是唯一知道 for 循环何时结束、何时迭代以及循环的内部结构是什么的人。

@DanielFarrell 我不同意，在 nodejs 中使用 Azure 消息代理的示例：import ServiceBusClient from '@azure/service-bus' export default async (config) => ServiceBusClient .createFromConnectionString(config.connection) .createQueueClient(config.queueName) .createReceiver(config.receiveMode) .registerMessageHandler(config.handler, config.errorHandler) azure sdk 负责在队列中进行“轮询”。当某些消息到达时，将调用处理程序。

当然，在异步 nodejs 领域，一切都是回调。嘿，如果你能找到对它的 SDK 支持，对于 Go SDK，太好了！我没看到，但也许我错过了！