Camel Azure 存储 Blob 使用者未按预期工作（Blob 持续消耗）答案

【问题标题】：Camel Azure Storage Blob consumer not working as expected (blob consumes continuously)Camel Azure 存储 Blob 使用者未按预期工作（Blob 持续消耗）
【发布时间】：2021-12-11 17:44:20
【问题描述】：

我已经编写了一个简单的路线来使用骆驼（版本 3.12.0）中 Azure Blob 存储中的 blob，如下所示：

 from("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&serviceClient=#client")
                .to({{filePath}});

blob 会不停地消耗，并且路由不会停止。有关如何在使用 blob 存储时实现幂等性的任何意见？

【问题讨论】：

您可以从存储中删除 blob，也可以将副本上传到已处理的“文件夹”并在处理完 blob 后删除原始文件。 “文件夹”是指为 blobName 添加前缀，即processed/test.json。您可以为此使用操作uploadBlockBlob 和operation。
感谢@PasiÖsterman。如果我们没有其他出路，这是我认为会尝试的一种选择。 :)
@PasiÖsterman - 我已经尝试使用这些选项上传到已处理，路线连续运行，并且文件不会上传到已处理文件夹，而是将文件上传到同一个容器：.to( "azure-storage-blob://{{camelazure}}/{{container1}}?blobName=processed/test.json&operation=uploadBlockBlob&serviceClient=#client");。还尝试使用 .to("azure-storage-blob://{{camelazure}}/{{container1}}/processed?blobName=test.json&operation=uploadBlockBlob&serviceClient=#client");你对此有什么意见吗？
是的，它会上传到同一个容器，但名称为processed/test.json，如果您希望将其存储到不同的容器中，则必须使用该容器的有效凭据为 URI 指定不同的容器。 Azure Blob 存储使用平面文件系统，因此没有真正的文件夹，只有基于 Blob 名称的虚拟文件夹。
就我个人而言，我更喜欢使用石英消费者端点和 azure-storage-blob 生产者端点从 blob 存储中获取 blob，然后对其进行处理，使用名称 processed/blobname-datetime 复制 blob，然后删除原始 blob。然后所有已处理的 blob 将位于同一个容器中，但位于以唯一 blob 名称命名的已处理虚拟文件夹中。

标签： apache-camel azure-blob-storage

【解决方案1】：

根据 Pasi Österman 的建议，当我们知道要下载和处理的 blob 名称时，以下解决方案运行良好。

from("timer://readBlob?fixedRate=true&period=60000")
    .to("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&operation=getBlob&serviceClient=#client")
    .process(exchange -> {
        InputStream inputStream = exchange.getMessage().getBody(InputStream.class);
        exchange.getIn().setBody(IOUtils.toString(inputStream, StandardCharsets.UTF_8.name()));
    })
    .to("file:{{filePath}}")
    .to("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=processed/test.json&operation=uploadBlockBlob&serviceClient=#client")
    .to("azure-storage-blob://{{camelazure}}/{{container1}}?blobName=test.json&operation=deleteBlob&serviceClient=#client");

【讨论】：