java - Apache camel 和 SQS 扩展客户端库

Question

关闭。这个问题需要更多 focused .它目前不接受答案。












想改进这个问题？更新问题，使其仅关注一个问题 editing this post .

5年前关闭。




Improve this question




所以。如您所知，AWS Simple Queue Service 对消息的大小有一些限制。还有一个library称为扩展客户端库，可帮助我们通过使用 S3 存储桶存储来避免此限制。我也用Apache Camel管理我的队列。

我想我只是误解了这个概念，但是我有问题让 Apache Camel 和扩展 SQS 库一起工作。

所以首先我定义了一条新路线

from("aws-sqs://queue?amazonSQSClient=#sqsClient" +
            "&maxMessagesPerPoll=10" +
            "&deleteAfterRead=false" +
            "&concurrentConsumers=10" +
            "&attributeNames=All" +
            "&messageAttributeNames=All"
        ).log(body());

sqsClient在 camel-context.xml 中定义

    <bean name="sqsClient" class="com.amazonaws.services.sqs.AmazonSQSAsyncClient">
    <constructor-arg>
        <bean class="com.amazonaws.auth.BasicAWSCredentials">
            <constructor-arg value="${access.key}"/>
            <constructor-arg value="${secret.key}"/>
        </bean>
    </constructor-arg>
    <property name="region" ref="awsRegion"/>
</bean>

而这一点在处理高达 256 kb 的消息时没有问题。

让我们更进一步。扩展 SQS 库检测消息是否大于 256 kb，并引用 S3 存储中的文件。稍后简单获取它。

["com.amazon.sqs.javamessaging.MessageS3Pointer",{"s3BucketName":"test-bucket-ascelhem","s3Key":"b022dfd8-ed40-4213-a41e-5a0e82090ef4"}]

在客户端能够使用它之前有一些预先配置:

首先，我们设置启用大负载支持的 SQS 扩展客户端配置。

ExtendedClientConfiguration extendedClientConfig = new ExtendedClientConfiguration().withLargePayloadSupportEnabled(s3, s3BucketName);

最后

AmazonSQS sqsExtended = new AmazonSQSExtendedClient(new AmazonSQSClient(credentials), extendedClientConfig);
        Region sqsRegion = Region.getRegion(Regions.US_WEST_1);
        sqsExtended.setRegion(sqsRegion);

那么我可以通过 Spring DSL 配置它还是应该获取消息并在处理器内部获取消息正文？

接下来，它是一个帮助我们获取消息体的方法。

    ReceiveMessageRequest receiveMessageRequest = new ReceiveMessageRequest(myQueueUrl);
    List<Message> messages = sqsExtended.receiveMessage(receiveMessageRequest).getMessages();

Apache Camel 是否以某种方式覆盖getMessage？因为在应用程序内部我们可以通过调用方法 body() 来获取消息正文。 .

如果有机会以这种方式配置 Apache Camel 以便它可以处理非常大的消息，那么行 new ExtendedClientConfiguration()
.withLargePayloadSupportEnabled(s3, s3BucketName);让我很困惑，因为我不知道如何在 xml 文件中调用静态方法。

最后一行

<bean name="sqsClient" class="com.amazonaws.services.sqs.AmazonSQSAsyncClient">

这里我们使用 AmazonSQSAsyncClient ，但正如我之前提到的，扩展客户端库通过 AmazonSQS 初始化... 界面？所以，我完全被困住了。

我也找到了 example使用这个库。它可能对你有帮助。

谢谢。

Answer 1

因此，我发现的最佳解决方案是创建一个进程，首先从 s3 对象获取消息正文并对其进行处理。