Camel 3.12.0 Azure Blob消费者持续重复消费与幂等性实现咨询

问题根因 #

当前路由未配置已消费Blob的过滤或标记规则，Camel的azure-storage-blob组件默认以轮询模式拉取Blob资源，只要目标Blob在源容器中存在且未被排除，每次轮询都会重新拉取并执行消费逻辑，因此会出现不间断重复消费的问题。

幂等性实现方案 #

你可以根据业务场景选择以下任意一种方案解决问题：

• 方案1：使用Camel内置幂等消费者组件（通用幂等方案）
  该方案通过维护已消费Blob的唯一标识库，自动过滤重复消费请求，对业务侵入性最低：

  // 示例使用内存幂等仓库，生产环境建议替换为JDBC、Redis等持久化实现，避免服务重启后丢失消费记录
  IdempotentRepository idempotentRepo = new MemoryIdempotentRepository();
  
  from("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&serviceClient=#client")
      // 可选择Blob的ETag、名称+最后修改时间作为唯一幂等Key
      .idempotentConsumer(header("CamelAzureStorageBlobETag"), idempotentRepo)
      .to({{filePath}});
  

• 方案2：消费后自动删除/移动Blob（源头避免重复拉取）
  如果业务允许消费后的Blob不需要保留在源容器，可以直接配置组件参数，消费完成后自动删除或归档Blob：

  // 消费完成后自动删除源Blob
  from("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&serviceClient=#client&deleteAfterRead=true")
      .to({{filePath}});
  
  // 或者消费完成后移动到归档容器，保留原始Blob数据
  from("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&serviceClient=#client&moveToAfterRead=archive-container")
      .to({{filePath}});
  

• 方案3：通过Blob元数据标记消费状态
  如果不允许删除或移动源Blob，可以在消费完成后给Blob添加自定义元数据作为已消费标记，拉取时过滤已标记的Blob：

  // 配置自定义filter过滤已标记为已消费的Blob
  from("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&serviceClient=#client&filter=#customBlobFilter")
      .to({{filePath}})
      // 消费完成后添加已消费元数据标记
      .setHeader("CamelAzureStorageBlobMetadata", constant(Map.of("consumed", "true")))
      .to("azure-storage-blob://{{camelazure}}/{{container1}}?operation=setBlobMetadata&serviceClient=#client");
  

  你需要自行实现customBlobFilter Bean，拉取Blob时判断元数据中是否存在consumed=true标记，存在则跳过消费。

内容的提问来源于stack exchange，提问作者DivyaShailesh