具有事件中心触发器的Azure功能写入重复消息

问题描述

我具有带有事件中心触发器的Azure功能。该集线器从设备接收消息并将其存储在Blob中。最近，我注意到重复的消息存储在blob中。 Blob存储区中的文件按上次修改日期排序，如果查看屏幕截图，您会发现情况并非如此。有人看过这个问题吗？

我还有一个Azure函数正在写入cosmos DB，对于blob中的重复消息，cosmos中没有相应的重复消息。

我还了解了时间序列见解，也没有重复的消息。

我打开了事件中心捕获，那里也没有重复的消息。

这是屏幕截图。

第一列是事件中心的排队时间的unix时间戳。如果我没有与文件名关联的GUID，它将引发异常。这是一个将数据存储在blob中的代码段。

dynamic msg = JObject.Parse(myEventHubMessage);
string deviceId = msg.deviceId;
if (deviceId == "5Y.....")
{
           var filename = "_" + ((DateTimeOffset)enqueuedTimeUtc).ToUnixTimeSeconds() + "_" + Guid.NewGuid().ToString() + ".json";
        
           var containerName = "containerName/";
        
           var path = containerName + deviceId + "/" + filename;
        
           using (var writer = binder.Bind<TextWriter>(new BlobAttribute(path)))
           {
                writer.Write(myEventHubMessage);
           }
 }

这里的逻辑非常简单。如果事件到达事件中心，则会触发该函数并将其存储在Azure Blob中。

解决方法

一个重要的要求是事件中心具有at-least-once delivery guarantee；强烈建议您以适合您的应用程序场景的方式，确保处理对事件重复具有弹性。

关于在这种情况下看到的重复，Azure Functions的绑定利用EventProcessorHost来读取事件并触发函数代码的执行。随着Azure功能自动按比例放大和缩小，EventProcessorHost的实例将加入并离开使用者组来负责处理配置的事件中心。

当处理器启动时，它将尝试与在同一使用者组中处于活动状态的其他处理器进行平衡处理工作。如果某个处理器无法通过声称拥有无用的分区来达到其应有的工作份额，则它将尝试从其他处理器中窃取分区的​​所有权。在此期间，新所有者将开始从最后记录的检查点开始读取。同时，旧所有者可能会将上次读取的事件分发给处理程序进行处理；除非尝试从事件中心服务读取下一组事件，否则它将不了解所有权已更改。当处理器关闭并放弃其分区所有权时，也会发生类似的情况。

因此，您将看到一些在启动或停止处理器时正在处理的重复事件，这些事件在处理器达到负载平衡的稳定状态时将消失。该窗口的持续时间应该很短，但是根据处理器的配置和所使用的检查点策略的不同而有所不同。