问题标签 [spring-integration-aws]

我正在批处理模式下使用消息。我想每 250 毫秒从流中提取 8 条消息。

我已经将大约 100 条消息推送到流中，并且我启动了消费者。

根据配置，我应该每 250 毫秒提取一次消息。但是轮询器不是每 250 毫秒拉一次消息。

我什至根本不处理任何东西。它只是日志。

每条消息之间的时间超过 250 毫秒。我有什么遗漏吗。

根据 Spring Aws Kinesis Binder 中的文档，readCapacity 和 writeCapacity 的默认值为 1

https://github.com/spring-cloud/spring-cloud-stream-binder-aws-kinesis/blob/master/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc#锁定注册表

readCapacity DynamoDb 表的读取容量。请参阅 Kinesis 预置吞吐量

默认值：1

writeCapacity DynamoDb 表的写入容量。请参阅 Kinesis 预置吞吐量默认值：1

从 Kinesis Client 代码中，我可以看到默认值为 10。

https://github.com/awslabs/amazon-kinesis-client/blob/master/amazon-kinesis-client/src/main/java/software/amazon/kinesis/leases/dynamodb/TableConstants.java

Spring Kinesis Binder 中是否有任何自定义。

编辑

我有读写能力为 40 的 locks 表。

我像这样配置了我的活页夹

我有三个使用这些配置运行的容器。

我看到针对 customLocks 表的 ProvisionedThroughputExceededException。

不确定活页夹是否试图使发电机数据库锁表过载。

我在 int-aws:s3-outbound-gateway 和 int-aws:s3-outbound-channel-adapter 都面临一个问题。问题是，假设如果我没有已配置为目标存储桶的存储桶访问权限，适配器应该在控制台中引发运行时错误，但我在控制台中没有收到任何错误，并且文件没有移动到各自的目的地。你能就此提出建议吗

我试图了解SqsMessageDrivenChannelAdapter解决内存问题的行为。上游系统转储了数千条消息aws-sqs-queue，所有消息都被立即接收SqsMessageDrivenChannelAdapter。在 AWS 控制台上，我看不到队列中的任何可用消息。然后SqsMessageProcesser每 5 秒处理 1 条消息。

这是日志：

2019-05-21 17:28:18 INFO SQSMessageProcessor:88 - --- sqsMessageProcesser 内部--- 2019-05-21 17:28:23 INFO SQSMessageProcessor:88 - --- sqsMessageProcesser 内部--- 2019-05- 21 17:28:28 INFO SQSMessageProcessor:88 - --- sqsMessageProcesser 内部--- 2019-05-21 17:28:33 INFO SQSMessageProcessor:88 --- sqsMessageProcesser 内部--- 2019-05-21 17:28 :38 INFO SQSMessageProcessor:88 - --- 在 sqsMessageProcesser 内--- ...................

这是否意味着在SqsMessageProcesser每 5 秒处理 1 条消息时，数千条消息被保存在 ? 的（服务器）内存中in-channel？

每个数据库事务大约需要 5 秒，目前我们在 PRD 上面临“内存不足”问题。

如果我在QueueChannelandsetMaxNumberOfMessages上设置容量会有帮助SqsMessageDrivenChannelAdapter吗？如果是，是否有计算这些值的标准方法？

如果不使用系统级别的默认配置，我无法在 Spring Cloud Stream Kinesis binder (1.2.0.RELEASE) 中连接到 AWS kinesis。仅当系统已配置为使用默认配置文件并且访问密钥 ID 和秘密访问密钥设置为使用 [默认] 配置文件时，应用程序才能正常工作。否则，它无法通过抛出此异常连接到 AWS 资源：

我尝试了以下选项，但没有一个对我有用：

• 设置环境变量AWS_ACCESS_KEY_ID和AWS_SECRET_ACCESS_KEY
• 设置 Java 系统属性aws.accessKeyId和aws.secretKey
• 设置cloud.aws.credentials.accessKey并cloud.aws.credentials.secretKey在application.yml文件中。

***强文本***我想在单个集成流程中使用 2 个 aws 出站适配器。一个是出站 S3 适配器和一个出站 SQS 适配器。我必须将文件从 smb 共享移动到 S3 存储桶，然后使用转换器，然后使用 SQS 适配器将转换后的消息发送到 SQS 队列。我可以通过 2 个集成流程来实现这一点，但我只想通过一个集成流程来实现这一点。如果我添加一个流的两个出站适配器部分，则只有其中一个在工作

我有一个 AWS S3 存储桶，我每周都会在其中放置一个新的 ZIP 文件。

我想在我现有的 Web 服务中添加一个功能，用 Spring Boot 编写：在本地同步存储桶并观察变化。

目前，同步效果很好：每当将新文件添加到存储桶中时，它都会在本地下载。但是，我不知道要监听文件更新，这是一种在本地下载新文件时触发的方法。可以做到吗？

这是我的一段代码：

我目前正在评估在新项目中使用 spring cloud stream kinesis binder 的可能性，但我遇到了一些问题。

当我启用 kcl-kpl-enabled: false 时，一切正常。但是，当我启用 kcl-kop 时，我不断收到以下错误：

这是我目前的配置：

我在测试中使用的依赖项的版本是：

我尝试使用spring-integration-aws从 S3 存储桶进行轮询来触发弹簧批处理作业。我的 S3 存储桶没有托管在 Amazon 上，它位于本地 minio 服务器上，所以我有一个自定义配置：

我以这种方式定义了我的 IntegrationFlow：

问题是，当轮询发生时，我收到以下错误：

发生这种情况是因为收到文件时，一些标头设置在spring-integration-aws：

AbstractRemoteFileStreamingMessageSource.java

由于getHostPort()方法REMOTE_HOST_PORT设置了标头。S3Session.java中的getHostPort ()然后调用getRegion()方法。

中的getRegion()方法不检查用户是否在字段AmazonS3Client中设置了值。signing region它只检查主机是否匹配“amazonaws.com”模式。

如何使用自定义端点配置从 S3 进行轮询？为什么getHostPort()方法不检查签名区域值？有可能解决这个问题吗？

我正在IntegrationFlow以这种方式使用 DSL 语法定义从 SFTP 到 S3 的流：

它按预期工作：文件很好地上传到我的 S3 存储桶。但是，我想避免SPEL语法，并将消息中的标头注入s3uploadMessageHandler方法，这样我可以使用简单的方法在方法ValueExpression中设置。为此，我改变了keyExpressions3UploadMessageHandler

到

但是现在这个处理程序似乎不再被触发了。日志中没有错误，我从日志中知道 SFTP 轮询仍在工作。

我试图找到这背后的原因，我发现当在 中输入句柄方法时IntegrationFlowdefinition.java，messageHandler类类型不同：它是一个S3MessageHandler没有 lambda 的MyCallingClass$lambda调用时，一个使用 lambda 表达式调用的时候。

我错过了什么让我的场景工作？