问题标签 [alpakka]

我正在尝试使用 akka-streams 和 akka-http 以及 alpakka 库将文件下载/上传到 Amazon S3。我看到两个可能相关的问题......

• 我只能下载很小的文件，最大的一个8kb。

• 我无法上传更大的文件。它失败并显示消息

  处理请求时出错：“子流源未在 5000 毫秒内实现”。完成 500 内部服务器错误响应。要更改默认异常处理行为，请提供自定义 ExceptionHandler。akka.stream.impl.SubscriptionTimeoutException：子流源未在 5000 毫秒内实现

这是我的路线

尝试下载 100KB 的文件最终会获取文件的截断版本，通常大小约为 16-25Kb 任何帮助表示赞赏

编辑：对于下载问题，我接受了 Stefano 的建议并得到了

这使它工作

我在这里阅读 Alpakka Cassandra 的文档

它使使用 Cassandra 作为源和接收器变得非常容易。但是流量使用情况如何。

通过流使用，我的意思是我没有将 Cassandra 用作源或接收器。但要查找数据。

使用Alpakka有可能吗？还是我必须自己编写 Cassandra jdbc 代码？

我写了以下Alpakka代码

我在 src/main/resources 文件夹中有一个名为“countrycapital.csv”的文件。

我的目标是为这个文件构建一个 Akka Streams Source 并处理这个文件的所有记录。

当我运行这段代码时。列表行引发异常

我在 build.sbt 中导入了以下库

假设我有使用 Akka（可能使用 Alpakka 和/或 Lagom）的应用程序。

我还能从像 Red Hat JBoss Fuse 这样的 iPaaS 中受益吗？

如果是，那有什么好处？如果不是，那如何称为作为 iPaaS 替代品的参与者系统的一部分？

我在 S3 上存储了 Kryo 序列化的二进制数据（数千个序列化对象）。

Alpakka允许将内容读取为data: Source[ByteString, NotUsed]. 但是 Kryo 格式不使用分隔符，因此我无法将每个序列化对象拆分为单独的ByteStringusing data.via(Framing.delimiter(...)).

因此，Kryo 实际上需要读取数据以了解对象何时结束，而且它看起来不适合流式传输。

是否有可能以流媒体方式实现这种情况，以便我Source[MyObject, NotUsed]在一天结束时完成？

我正在编写一些代码来将数据插入 MySQL。我决定使用 Alpakka Slick Connector。使用该示例：https ://developer.lightbend.com/docs/alpakka/latest/slick.html#using-a-slick-flow-or-sink我已成功将数据添加到数据库，但它需要我通过带有 SQL 命令的字符串到 Sink。由于我收集了案例对象，因此我需要从这些对象构建命令。我可以向案例类添加生成 INSERT、UPDATE 和 DELETE 命令的方法，但也许有一些更好的方法可以添加整个对象集合，而无需使用 SQL 包装数据。

我正在使用Alpakka它的JMS连接器来将数据从Oracle AQ. 我可以按照本指南提出以下非常基本的实现。

我的问题是如何使它具有事务性，这样我就可以保证如果抛出异常我的消息不会丢失。

如果是PL/SQL，解决方案将是这样的：

我正在使用Alpakka JMS连接器从Oracle AQ中取出项目，并使用Slick进行数据库操作，队列和表位于同一架构上。

我需要以事务方式进行出队和数据库操作，所以如果一个失败，另一个也会失败。有没有办法实现这个功能？

到目前为止我想出的是有两个单独的交易，这是不安全的：

在每条消息成功发送到 JMS 或失败后如何进行回调？

更具体的例子

客户将项目发送到sourceQueue：

val result是插入项目的结果，sourceQueue并不意味着它已经发送到JMS。当项目通过接收器过程并插入到 JMS 队列时，我需要触发一个事件。

我有一个场景，我正在使用 alpakka 启动多个 jmsSource（用于不同的队列）。我还需要在任何时间点分离队列。所以我在 jms akka 流中添加了 KillSwitch，如下所示：-

.. 等等。

启动应用程序后，所有队列都已连接并且工作正常。但是当我尝试使用 stop 方法分离队列时，并非所有队列都断开连接并且行为是随机的。我还检查了所有听众的 killSwitch 是否不同。

有人可以告诉我这里出了什么问题吗？