9.5. 使用方法
例如,若要从 storageAccount
存储帐户中的 queue messageQueue
获取消息内容,并使用以下代码片段:
from("azure-storage-queue://storageAccount/messageQueue?accessKey=yourAccessKey"). to("file://queuedirectory");
9.5.1. 由组件制作者评估的消息标头
标头 | 变量名称 | 类型 | 操作 | 描述 |
---|---|---|---|---|
|
|
|
| 列出队列的选项 |
|
|
| All | 将引发 \{@link RuntimeException} 之外的可选超时值。 |
|
|
|
| 与队列关联的元数据 |
|
|
|
| 邮件将在队列中显示多久。如果未设置,则值将默认为 7 天,如果传递了 -1,则消息不会过期。生存时间必须是 -1 或任何正数。 |
|
|
|
| 消息在队列中不可见的超时时间。如果未设置,则默认值为 0,则消息将立即可见。超时必须在 0 秒到 7 天之间。 |
|
|
|
|
当设置为 |
|
|
|
| 必须匹配的唯一标识符,才能删除或更新消息。 |
|
|
|
| 要删除或更新的消息的 ID。 |
|
|
|
| 如果队列中存在的消息少于请求的所有消息,则要获取的最大消息数。如果只检索空白的 1 消息,则允许的范围为 1 到 32 个消息。 |
|
|
| All | 指定要执行的制作者操作,请参阅此页面中的与制作者操作相关的 doc。 |
|
|
| All | 覆盖队列名称。 |
9.5.2. 由组件制作者或消费者设置的消息标头
标头 | 变量名称 | 类型 | 描述 |
---|---|---|---|
|
|
| 发送到队列的消息 ID。 |
|
|
| 消息插入到队列中的时间。 |
|
|
| 消息将过期并自动删除的时间。 |
|
|
| 删除/更新消息需要这个值。如果删除操作失败,则消息被另一个客户端取消排队。 |
|
|
| 在 Queue 中,消息再次变为可见的时间。 |
|
|
| 消息已排队的次数。 |
|
|
| 返回可由用户使用的非解析的 httpHeaders。 |
9.5.3. 高级 Azure Storage Queue 配置
如果您的 Camel 应用程序在防火墙后面运行,或者需要对 QueueServiceClient
实例配置拥有更多控制,您可以创建自己的实例:
StorageSharedKeyCredential credential = new StorageSharedKeyCredential("yourAccountName", "yourAccessKey"); String uri = String.format("https://%s.queue.core.windows.net", "yourAccountName"); QueueServiceClient client = new QueueServiceClientBuilder() .endpoint(uri) .credential(credential) .buildClient(); // This is camel context context.getRegistry().bind("client", client);
然后,在 Camel azure-storage-queue
组件配置中引用这个实例:
from("azure-storage-queue://cameldev/queue1?serviceClient=#client") .to("file://outputFolder?fileName=output.txt&fileExist=Append");
9.5.4. 在注册表中自动检测 QueueServiceClient 客户端
组件能够检测 QueueServiceClient bean 在注册表中的存在。如果是该类型的唯一实例,它将用作客户端,并且不必将它定义为 uri 参数,如上例所示。这可能对端点的更智能配置非常有用。
9.5.5. Azure Storage Queue Producer 操作
Camel Azure Storage Queue 组件在制作者端提供广泛的操作:
服务级别的操作
对于这些操作,需要 accountName
。
操作 | 描述 |
---|---|
| 列出存储帐户中从指定标记开始传递过滤器的队列。 |
队列级别的操作
对于这些操作,需要 accountName
和 queueName
。
操作 | 描述 |
---|---|
| 创建新队列。 |
| 永久删除队列。 |
| 删除队列中的所有消息。 |
|
默认 Producer Operation Sends 消息带有给定时间到实时,以及消息在队列中不可见的超时时间。消息文本从交换消息正文评估。默认情况下,如果队列不存在,它将首先创建一个空队列。如果要禁用此功能,请将 config |
| 删除队列中的指定消息。 |
| 从队列检索最多消息数量,并在超时时间内从其他操作隐藏它们。但是,由于可靠性的原因,它将不会从队列中取消排队。 |
| 从队列前进到最大消息数。 |
| 使用新消息更新队列中的特定消息并重置可见性超时。消息文本从交换消息正文评估。 |
请参阅此页面中的示例部分,了解如何在您的 camel 应用程序中使用这些操作。
9.5.6. 消费者示例
要将队列消耗到一个批处理中最多 5 个消息的文件组件,请执行如下操作:
from("azure-storage-queue://cameldev/queue1?serviceClient=#client&maxMessages=5") .to("file://outputFolder?fileName=output.txt&fileExist=Append");
9.5.7. 制作者操作示例
-
listQueues
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g, to only returns list of queues with 'awesome' prefix: exchange.getIn().setHeader(QueueConstants.QUEUES_SEGMENT_OPTIONS, new QueuesSegmentOptions().setPrefix("awesome")); }) .to("azure-storage-queue://cameldev?serviceClient=#client&operation=listQueues") .log("${body}") .to("mock:result");
-
createQueue
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: exchange.getIn().setHeader(QueueConstants.QUEUE_NAME, "overrideName"); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=createQueue");
-
deleteQueue
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: exchange.getIn().setHeader(QueueConstants.QUEUE_NAME, "overrideName"); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=deleteQueue");
-
clearQueue
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: exchange.getIn().setHeader(QueueConstants.QUEUE_NAME, "overrideName"); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=clearQueue");
-
sendMessage
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: exchange.getIn().setBody("message to send"); // we set a visibility of 1min exchange.getIn().setHeader(QueueConstants.VISIBILITY_TIMEOUT, Duration.ofMinutes(1)); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client");
-
deleteMessage
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: // Mandatory header: exchange.getIn().setHeader(QueueConstants.MESSAGE_ID, "1"); // Mandatory header: exchange.getIn().setHeader(QueueConstants.POP_RECEIPT, "PAAAAHEEERXXX-1"); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=deleteMessage");
-
receiveMessages
:
from("direct:start") .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=receiveMessages") .process(exchange -> { final List<QueueMessageItem> messageItems = exchange.getMessage().getBody(List.class); messageItems.forEach(messageItem -> System.out.println(messageItem.getMessageText())); }) .to("mock:result");
-
PaesMessages
:
from("direct:start") .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=peekMessages") .process(exchange -> { final List<PeekedMessageItem> messageItems = exchange.getMessage().getBody(List.class); messageItems.forEach(messageItem -> System.out.println(messageItem.getMessageText())); }) .to("mock:result");
-
updateMessage
:
from("direct:start") .process(exchange -> { // set the header you want the producer to evaluate, refer to the previous // section to learn about the headers that can be set // e.g: exchange.getIn().setBody("new message text"); // Mandatory header: exchange.getIn().setHeader(QueueConstants.MESSAGE_ID, "1"); // Mandatory header: exchange.getIn().setHeader(QueueConstants.POP_RECEIPT, "PAAAAHEEERXXX-1"); // Mandatory header: exchange.getIn().setHeader(QueueConstants.VISIBILITY_TIMEOUT, Duration.ofMinutes(1)); }) .to("azure-storage-queue://cameldev/test?serviceClient=#client&operation=updateMessage");
9.5.8. 开发备注(Important)
在此组件上开发时,您需要获取 Azure accessKey 来运行集成测试。除了模拟的单元测试外,还需要针对您进行的每个更改运行集成测试,甚至客户端升级,因为 Azure 客户端即使在次版本升级时也会出现问题。要运行集成测试,在这个组件目录中运行以下 maven 命令:
mvn verify -PfullTests -DaccountName=myacc -DaccessKey=mykey
其中,by accountName
是 Azure 帐户名称,accessKey
是从 Azure 门户生成的访问密钥。