问题标签 [spring-integration-aws]

我正在寻找一个使用 Spring Integration + Amazon SQS 服务接收消息并将消息发送到队列的 Spring 应用程序的工作示例。

我试图为 Aws kinesis 流消费者实现负载平衡

根据我正在尝试实施的文档

我有3 个容器，如果需要，我想在不重新启动现有容器的情况下启动新容器（最多 6 个） 。

1. instanceIndex 从 0 或 1 开始。
2. 如果我将 instanceCount 设为 6，但只启动了三个实例，那么在我启动新实例之前，是否会消耗所有消息。
3. 在文档中，有一个名为 spring.cloud.stream.bindings..consumer.concurrency 的属性，你能帮助它的重要性。
4. 由于某些原因，如果任何一个实例出现故障，是否有任何消息未被消费。

你能帮助我们吗

我正在使用 1 条碎片推送 100 条消息。

我正在循环推送消息以进行测试

我正在使用如下消息

消息消费无序。

我是不是错过了什么。

我正在将 Spring Integration 与 AWS SQS 队列集成。

当我用注释的方法 @ServiceActivator引发异常时，我遇到了问题。在这种情况下，消息似乎无论如何都会从队列中删除。我已配置MessageDeletionPolicy为ON_SUCCESSin SqsMessageDrivenChannelAdapter。

这是我的频道/适配器配置 https://github.com/sdusza1/spring-integration-sqs/blob/master/src/main/java/com/example/demo/ChannelConfig.java

我已经尝试使用@SqsListener注释做同样的事情，并且消息没有按预期删除。

我在这里创建了一个迷你 Spring Boot 应用程序来演示这个问题： https ://github.com/sdusza1/spring-integration-sqs

请帮忙 ：）

我正在使用批处理模式从运动流中提取记录。我们正在使用 spring aws kinesis binder。

大多数时候，我们无法从流中提取消息。只有有时我们能够从流中提取消息。

我的配置如下所示

我的配置

当我启用调试日志记录时，我可以看到这个异常

我尝试将批量大小减少为 150，并将 idleBetweenPools 减少到 1 秒。我还将 readCapacity 和 writeCapacity 更新为 10。但同样的错误。

从 AWS 控制台，我可以看到 SpringIntegrationLockRegistry 已超过读取阈值。

你能帮助我们了解什么是错的。

它有时有效，有时无效。

我正在使用 spring kinesis binder 以批处理模式消费消息

有时我无法使用流中的消息。这并非一直发生

[ShardConsumer{shardOffset=KinesisShardOffset{iteratorType=AFTER_SEQUENCE_NUMBER, sequenceNumber='49594358705006691330463332232285735104253344290967650306', 时间戳=null, 流='mystream-1', shard='00000}']=false}的记录在序列号 [null] 上。暂停消费 [1000] 毫秒。

我不断收到此消息，并且消息没有被消耗。

我的 Conf 如下所示

流中有消息，但我不会间歇性地消费。你能帮忙吗？

我的应用程序从一个流中消费，然后将消息推送到三个流上

粘合剂：

配置：

我们推送的数据流在每条记录中大约 800kb。我们看到 AbstractAwsMessageHandler/AmazonKinesisAsyncClient 中有更多数据导致非常频繁的 GC 刷新。

我们使用的是 1.0.0.RELEASE 版本的 Binder

你能帮忙吗？

我在使用SqsMessageDrivenChannelAdapter时遇到了一些非常糟糕的吞吐量问题，这是围绕's 的spring-integration-awsSpring Integration 抽象。spring-cloud-awsSimpleMessageListenerContainer

问题似乎是SimpleMessageListenerContainer一次只能请求 10 条消息（AWS 限制），并且使这些请求非常慢 - 特别是，我观察到 ~15 tps，这太慢了。

这是我的代码：

receivedSqsMessageChannel是最终在消息Acknowledgement#acknowledge()被调用时结束的流的输入。

SimpleMessageListenerContainer鉴于（因此）还没有任何功能SqsMessageDrivenChannelAdapter可以在多个线程上进行轮询（有关于 的讨论，spring-cloud-aws但距离实现还很遥远，之后spring-integration-aws将引入对它的支持），以及我现在拥有的 TPS是不可接受的，我认为目前最好的“修复”是让多个SqsMessageDrivenChannelAdapters 同时轮询并输出到receivedSqsMessageChannel.

我怎样才能在春天实现这一目标？有没有办法让我SqsMessageDrivenChannelAdapter使用 Spring 注释同时运行一个 s 池？

我想在发送到目标系统之前使用一些逻辑来验证文件内的数据（以保持数据完整性）。为此，我正在使用这样的文件及其校验和文件逻辑，abc.txt 是原始文件 abc.txt.checksum 是它的校验和文件。abc.txt.checksum 是 abc.txt 文件的哈希值，将用于在发送到目标系统之前进行验证。两个文件将位于同一源位置。我有一个过滤条件，在轮询时不选择 .checksum 文件，因为我不想将 .checksum 文件发送到目标系统。选择 abc.txt 文件后，我将使用一个 java 逻辑在内部创建此文件的哈希并将其存储到内存中，并且必须与已经存在的校验和（abc.txt.checksum）数据进行比较在那个位置。如果两个校验和数据都匹配，那么只有原始文件必须移动到目标系统，否则它不应该处理并抛出错误。有人可以指导我如何实现这一目标吗？或者我将如何获取 .checksum 文件 (abc.txt.checksum)，因为它始终存在于源位置？

我有一个要求，我想根据源位置的文件扩展名设置目标位置，例如，在源位置我有以下文件 abc.txt 、mno.txt、pqr.csv 和 xyz.pdf 。一旦适配器开始轮询，abc.txt 和 mno.txt 将移动 txtDesinationPath pqr.cv 将不得不移动 csvDestinationPath 和 xyz.pdf 将不得不分别移动 pdfDesinationPath。您能否建议我一些逻辑来实现这一点。