如何避免EventhubTriggered函数应用中的重复数据处理

public static class EventHubTriggeredFunction
{
    [FunctionName("EventHubTriggeredFunction")]
    public static async Task Run(
        [EventHubTrigger("eventhubname", Connection = "EventHubConnectionString")] string[] messages,
        ILogger log)
    {
        foreach (var message in messages)
        {
            var messageId = GetMessageId(message);
            if (!IsMessageProcessed(messageId))
            {
                await ProcessMessage(message);
                MarkMessageAsProcessed(messageId);
            }
        }
    }

    private static string GetMessageId(string message)
    {
        // 从消息中提取唯一标识符
        return Guid.NewGuid().ToString();
    }

    private static bool IsMessageProcessed(string messageId)
    {
        // 检查消息是否已经处理过
        // 可以使用数据库或缓存来存储已处理的消息标识符
        return false;
    }

    private static async Task ProcessMessage(string message)
    {
        // 处理消息的逻辑
    }

    private static void MarkMessageAsProcessed(string messageId)
    {
        // 标记消息为已处理
        // 可以使用数据库或缓存来存储已处理的消息标识符
    }
}

public static class EventHubTriggeredFunction
{
    [FunctionName("EventHubTriggeredFunction")]
    public static async Task Run(
        [EventHubTrigger("eventhubname", Connection = "EventHubConnectionString")] string[] messages,
        ILogger log)
    {
        using (var transaction = await BeginTransaction())
        {
            foreach (var message in messages)
            {
                var messageId = GetMessageId(message);
                if (!IsMessageProcessed(messageId))
                {
                    await ProcessMessage(message);
                    MarkMessageAsProcessed(messageId);
                }
            }
            await CommitTransaction(transaction);
        }
    }

    private static async Task<IDbTransaction> BeginTransaction()
    {
        // 开始事务
        return await Task.FromResult<IDbTransaction>(null);
    }

    private static async Task CommitTransaction(IDbTransaction transaction)
    {
        // 提交事务
    }
}

public static class EventHubTriggeredFunction
{
    [FunctionName("EventHubTriggeredFunction")]
    public static async Task Run(
        [EventHubTrigger("eventhubname", Connection = "EventHubConnectionString")] string[] messages,
        ILogger log)
    {
        foreach (var message in messages)
        {
            var messageId = GetMessageId(message);
            if (!await IsMessageProcessedAsync(messageId))
            {
                await ProcessMessage(message);
                await MarkMessageAsProcessedAsync(messageId);
            }
        }
    }

    private static string GetMessageId(string message)
    {
        // 从消息中提取唯一标识符
        return Guid.NewGuid().ToString();
    }

    private static async Task<bool> IsMessageProcessedAsync(string messageId)
    {
        // 检查消息是否已经处理过
        // 使用 Azure Cosmos DB 存储已处理的消息标识符
        return false;
    }

    private static async Task ProcessMessage(string message)
    {
        // 处理消息的逻辑
    }

    private static async Task MarkMessageAsProcessedAsync(string messageId)
    {
        // 标记消息为已处理
        // 使用 Azure Cosmos DB 存储已处理的消息标识符
    }
}