如何将最新事件与Azure Event Hubs Go SDK(azeventhubs)关联？

2cmtqfgy 于 2023-11-14 发布在 Go

关注(0)|答案(1)|浏览(112)

我正在从azure-event-hubs-go/v3迁移到更新的azeventhubsGo SDK。在旧版SDK中，有一个ReceiveOption参数允许我指定从何处开始消费事件。
在新的SDK中，我使用以下代码来初始化处理器：

processor, err := azeventhubs.NewProcessor(
    e.ConsumerClient, 
    checkpointStore, 
    &azeventhubs.ProcessorOptions{
        UpdateInterval: time.Second, 
        Prefetch: 0, 
        StartPositions: azeventhubs.StartPositions{
            Default: azeventhubs.StartPosition{
                Latest: to.Ptr(true), 
                EnqueuedTime: to.Ptr(time.Now()), 
                Inclusive: true
            }
        }
    }
)

字符集
但是，我注意到这些事件是从最后一个检查点而不是从最近发送的事件中消耗的。

尝试内容：我尝试了ConsumingEventsUsingConsumerClient和ConsumingEventsWithCheckpoints两个示例，但它们的行为方式相同，都是消费上一个检查点的事件，而不是最近的事件。
**预期：**我希望处理器开始消费设备发送的最新事件，设备每秒发送一条消息。如何使用azeventhubs Go SDK实现此行为？

来源：https://stackoverflow.com/questions/77401434/how-to-consume-the-latest-event-with-azure-event-hubs-go-sdk-azeventhubs

1条答案

按热度按时间

i7uq4tfw1#

我最初很难理解AMQP的基本机制，但是，我很高兴地报告，这个问题已经成功地解决了。

var wg sync.WaitGroup
wg.Add(1)
for _, partition := range p.PartitionIDs {
    go func(partition string) {
        defer wg.Done()
        partitionClient, err := consumerClient.NewPartitionClient(partition, nil)
        if err != nil {
            panic(err)
        }
        receiveCtx, cancel := context.WithTimeout(context.TODO(), time.Second*30)
        defer cancel()
        for {
            events, err := partitionClient.ReceiveEvents(receiveCtx, 1, nil)
            if err != nil && !errors.Is(err, context.DeadlineExceeded) {
                panic(err)
            }
            for _, evt := range events {
                fmt.Printf("partition: %s\n", partition)
                fmt.Printf("Body: %s\n", string(evt.Body))
            }
        }
    }(partition)
}
wg.Wait()

字符集
我感谢Azure客户支持服务团队提供的宝贵帮助。

展开查看全部

赞(0）回复(0）举报 2023-11-14

我来回答

如何将最新事件与Azure Event Hubs Go SDK(azeventhubs)关联？

1条答案

相关问题

热门标签

最新问答