在Akka Streams中压缩两个流
Akka Streams是一个用于构建可伸缩和高性能的流处理应用的工具包。它基于Actor模型和Reactive Streams规范，提供了一种声明式的方式来处理数据流，并可以方便地进行流操作和转换。
在Akka Streams中，压缩两个流是指将两个流合并为一个流，并通过对其中的元素进行一定的处理来减少数据的大小。这可以通过使用zip操作符来实现。zip操作符会将两个流中对应位置的元素进行压缩，并生成一个新的流，其中的每个元素是一个元组，包含了两个原始流中对应位置的元素。
压缩两个流的好处是可以减少传输的数据量，提高网络传输的效率。在一些场景中，比如实时数据传输、消息队列处理等，压缩流可以有效地减少网络带宽的使用和延迟。
在腾讯云中，可以使用腾讯云的流计算产品Tencent Cloud StreamCompute来实现压缩两个流。Tencent Cloud StreamCompute是一项完全托管的流式计算服务，提供了高可用性、低延迟的数据处理能力，可以方便地进行流处理、数据转换和聚合操作。
相关产品链接：Tencent Cloud StreamCompute
相关·内容
1回答
我很难理解NotUsed和Done在Akka流中的目的和意义。
让我们看看以下两个简单的例子：
使用NotUsed :
implicit val system = ActorSystem("akka-streams")
implicit val materializer = ActorMaterializer()
val myStream: RunnableGraph[NotUsed] =
Source.single("stackoverflow")
.map(s => s.toUpperCase())
.to(Sink.foreach(pr
浏览 4提问于2017-10-24得票数 5
回答已采纳
2回答
我有一个关于使用Akka Streams和Akka集群的问题。我正在尝试做一个使用Akka Streams和Akka Cluster的分布式字数统计版本。
我想构建一个Akka Streams客户端，它将文本文件作为流I/O读取，并将单词流发送到远程集群。这是客户端的代码：
final Path file = Paths.get("example.txt");
final Source<ByteString, CompletionStage<IOResult>> read = FileIO.fromPath(file);
final Source<
浏览 5提问于2018-11-10得票数 2
让我们假设我们有两个Akka Stream流，每个流都在自己的JVM上运行。
// A reactive streams publisher running on JVM 1:
val stringPublisher: Publisher[String] = Source(() => "Lorem Ipsum".split("\\s").iterator).runWith(Sink.publisher[String])
// A reactive streams subscriber running on JVM 2:
def subscriber: S
浏览 0提问于2015-03-06得票数 2
如何在argo和kubernetes中运行akka streams应用程序，我找到了关于kubernetes和akka集群的文档，但我不需要akka集群，我只需要运行一个具有许多参与者的短暂akka应用程序；还是akka集群是必要的？
浏览 29提问于2021-02-12得票数 0
回答已采纳
1回答
我是AKKA的新手(我使用的是java)。我不知道如何将zip文件传输到远程参与者。我应该使用akka-http吗？我尝试过akka-streams，但我不能理解如何将二进制文件传输到远程actors。
浏览 1提问于2017-09-19得票数 0
我正在使用Akka流从AWS S3桶读取.gz文件，应用程序读取数据块并对其进行处理。
当我从一开始读取文件时，下面的代码运行良好，但是如果我从特定的偏移量启动文件，则解压缩失败并抛出ParsingException。
代码：
object StreamApp extends App {
// Run 1
//startStreaming(0)
// Run 2
startStreaming(33339)
def startStreaming(pointer: Long): Unit = {
println("Stream A
浏览 0提问于2018-02-26得票数 1
2回答
我开始学习，它是一个处理具有背压功能的数据的框架。库是的一部分，它将自己描述为：
Akka是一个工具包和运行时，用于在JVM上构建高度并发、分布式和弹性强的消息驱动应用程序。
这些能力来自阿克卡演员的本性。然而，在我看来，流处理和参与者是无关的概念。
问题: Akka Streams是否利用了Akka演员的这些特点？如果是，你能解释演员如何帮助溪流吗？
浏览 2提问于2015-07-31得票数 4
回答已采纳
当我试图对响应进行两次解组时，我遇到了这个错误，这是因为我需要将响应作为case类，并且我还需要它作为json-string，因为对于对象，我在第一个流中使用它，在另一个第二个流中使用json-string，所以这就是为什么我需要这两个类。 错误日志： 21:08:16.174 [Extractor-akka.actor.default-dispatcher-14] DEBUG akka.actor.RepointableActorRef - Aborting tcp connection to api.clickup.com:443 because of upstream failure:
浏览 18提问于2021-04-07得票数 1
1回答
我在用可重新启动的源和接收器处理akka流中的错误。
object Main extends App {
implicit val system: ActorSystem = ActorSystem("akka-streams-system")
val restartSettings =
RestartSettings(1.seconds, 10.seconds, 0.2d)
val restartableSource = RestartSource.onFailuresWithBackoff(restartSettings) {() => {
浏览 8提问于2022-01-10得票数 1
回答已采纳
1回答
阅读akka的文档，我不太清楚消息的顺序，以及是否可以强制执行。让我用我为聊天服务器编写的一小部分代码来设置我问题的上下文。
def flowShape(user: User) = GraphDSL
.create(Source.actorRef[ChatMessage](bufferSize = 5, OverflowStrategy.fail)) {
implicit builder =>
implicit chatSource =>
import GraphDSL.Implicits._
val messageFromOut
浏览 4提问于2017-11-11得票数 0
回答已采纳
我正在尝试akka streams，但在我的简单示例中，我无法获得工作的反压力。无可否认，我没有使用akka (streams)的经验，所以我可能错过了一些重要的东西。
我生产(在队列中提供)整数的速度比消费它们的速度更快，所以我认为反压力会发挥作用。我的目标是始终使用放入队列的最新项(这就是为什么我在源队列上使用bufferSize =1和OverflowStrategy.dropHead()的原因)。
public class SimpleStream {
public static void main(String[] argv) throws InterruptedExcept
浏览 0提问于2017-05-28得票数 3
我在我的应用程序中使用akka流来处理实时数据。数据量如此之大，所以我想横向扩展我的应用程序。可以帮助我理解akka-streams是否支持集群吗？如果是，请分享例子。
浏览 1提问于2019-07-12得票数 0
我正在寻找akka.stream.scaladsl.Source构造方法，它允许我简单地从不同的代码位置发出下一个值(例如监视系统事件)。
我需要一些类似于Promise的东西。承诺向Future发出单个值。我需要向Source发出多个值。
就像monix.reactive.subjects.BehaviorSubject.onNext(_)
我不太关心背压。
目前，我已经使用monix & akka-streams (下面的代码)实现了这一点，但我希望应该只有akka-streams解决方案：
import akka.stream.scaladsl.{Flow,
浏览 4提问于2020-03-05得票数 1
回答已采纳
1回答
为了解决以下问题，我尝试使用akka-streams和akka-http：
我们有两个http客户端(A和B)发送请求到一个http服务器(C)。双方使用akka-http进行通信。
与B相比，来自A的请求具有更高的优先级，但这两个请求都应该是相同的进程。所以C应该先处理来自A的请求，B的请求是第二优先
当然，我们希望每一端都有背压。
为了将传入的连接合并到一个输出中，我提出了以下代码：
val g = RunnableGraph.fromGraph(FlowGraph.create() { implicit b: FlowGraph.Builder[Unit] =>
浏览 5提问于2015-12-16得票数 0
回答已采纳
1回答
我正在使用akka streams。因此，我有一个具有此功能的角色。
这两个消息有时都是按顺序处理的。但由于异步调用(我想)，我在这里得到了死信。
actor.tell(message,ActorRef.noSender());
actor.tell(PoisonPill.getInstance(),ActorRef.noSender())
有没有人能帮我确保以特定的顺序运行这段代码？
浏览 6提问于2019-07-29得票数 0
2回答
我不知道如何立即停止akka stream Runnable Graph？如何使用killswitch来实现这一点？我创建akka streams才几天。在我的例子中，我从文件中读取行，并在flow中执行一些操作，然后写入接收器。我想要做的是，在我想要的时候立即停止读取文件，我希望这可能会停止整个运行的图形。任何关于这方面的想法都将非常感谢。
提前谢谢。
浏览 9提问于2016-08-03得票数 5
回答已采纳
2回答
我目前正在尝试在Scala中使用具有管道和过滤器架构的Akka Streams编写程序。我有一个特定的图形，它应该接受一个输入并将其输出到多个流。最后，应该将所有不同流的结果合并为一个。在我的例子中，输入将是所有类型的tweet。然后这些推文首先转到不同的过滤器，所有过滤器都在一种类型上过滤，然后进行扫描，这只是简单地计算它看到了多少某种类型。在此之后，我希望输出是这些扫描的返回值，并将其组合到一个元组中。 现在，我已经为此设置了一个特定的图形DSL，它使用Broadcast和ZipWith来完成此任务。我的代码如下： val splitStreams =
Flow.fromGraph(
浏览 23提问于2019-12-21得票数 0
回答已采纳
我想做以下使用akka streams..我尝试广播，和alsoTo，但它没有解决我的问题
Source -> DBSink -> APISink。
我首先想把元素发送到DB，当事务成功完成后，我想把它发送到APISink..如果first Sink中的事务失败，我不想将其发送到API。
浏览 0提问于2020-06-11得票数 0
我有一个带有Akka流的管道，它使用自定义调度程序。
Akka流代码：
// Pulls in data via network requests
source
// Downstream will be sent to a different actor.
// Also, a "buffer" of 16 elements will be created between those actors.
.async
.via(writeToDatabase())
.toMat(Sink.ignore)(Keep.right)
.withAttribute
浏览 15提问于2021-09-03得票数 1
使用Akka-Http和Akka-Streams版本2.4，我如何构造一个流，以便它有一个接收器，即一个websocket。
例如，我希望有一个源生成随机数，这些数字被过滤，假设保持偶数，然后流到WebSocket作为接收器。
谢谢
浏览 0提问于2016-02-06得票数 1
回答已采纳
我一直在使用 for Akka.Net，并注意到试图将缓冲区和节流方法组合在一起时出现了这个错误：
using (var system = ActorSystem.Create("test-system"))
using (var materializer = system.Materializer(GetSettings(system)))
{
int index = 0;
var sink = Sink.ActorRefWithAck<KeyValue>(
system.ActorO
浏览 0提问于2018-10-10得票数 0
回答已采纳
1回答
目标：
我希望看到Akka的文件IO尽可能的异步和非阻塞。
我目前的知识状况：
在中，您可以阅读以下内容：
备注
由于当前版本的Akka (2.3.x)需要支持JDK6，所以当前提供的文件IO实现无法使用异步文件IO操作，因为这些操作是在JDK7 (和更新版本)中引入的。一旦Akka可以自由地要求JDK8 (来自2.4.x)，这些实现将被更新以使用新的NIO (即AsynchronousFileChannel)。
目前的akka版本是'2.5.4‘。目前版本的akka流是'2.11‘或'2.12’。在中，省略了abhove的注释，它只是明确地提到，文件IO意味着
浏览 3提问于2017-08-18得票数 0
回答已采纳
1回答
我正在使用akka-streams，我遇到了一个例外，因为我在akka-http上使用了最大的Http池。有一个列表元素的来源，这些元素被拆分，从而转换为SubFlows。SubFlows发出http请求。虽然我在SubFlow上放置了一个缓冲区，但该缓冲区似乎在每个SubFlow上生效。是否有一种基于对SubFlows生效的源的缓冲区？
浏览 4提问于2020-09-11得票数 0
回答已采纳
这个问题已经在Akka Streams和Akka HTTP的旧版本中被问到了。这个答案的问题是，对于每个流，你需要为每个请求物化一个新的流，这并不便宜。
一年后，在最新版本中，将流与路由DSL集成的最佳方式是什么？
同样，我有一个想要添加到路由的复杂Flow，如果我将流映射到Flow[HttpRequest, HttpResponse, _]，我可以使用此流完成路由吗？有没有更好的方法或者是使用低级API的唯一方法？
如果是这样的话，做路由映射的最好方法是什么？
浏览 0提问于2017-05-23得票数 2
2回答
我有一个现有的akka应用程序构建在socko websockets上。与套接字的通信在单个参与者内部进行，离开和进入参与者的消息(分别为传入消息和传出消息)被标记为套接字id，该套接字id是socko websocket的第一类属性(在socko中，连接请求到达带有id标记的连接请求，所有生命周期转换(例如握手、断开连接、传入帧等)都有类似的标记)。
我想使用akka-http重新实现这个单独的参与者(目前，由于明显的原因，socko或多或少是放弃软件)，但这并不简单，因为这两个库在概念上非常不同；akka-http隐藏握手、断开连接等的较低级别的细节，只需发送任何一个参与者绑定到http服
浏览 0提问于2016-05-26得票数 2
回答已采纳
1回答
我对Akka streams技术非常陌生，我没有找到任何关于我的用例的信息来源。
我希望使用Akka流来使用Kafka主题中的消息，处理它们(同时使用另一个数据源)，并将输出存储在mongo集合中。
为了实现并行性，我们正在研究一个主题的方法(每个组织都有一个分区)和集合。主题和集合可以动态创建，应用程序需要具有高度的可伸缩性，并部署在多个荚上。
实现这种行为的最佳实践是什么？有可能用阿克卡流吗？谢谢!
浏览 3提问于2022-03-16得票数 1
如何从Akka Streams的Sink中抛出的异常中恢复？
简单的例子：
<code>A0</code>
输出：
<code>A1</code>
如何处理异常并从源代码转到下一个元素？(又名5,6,7,8,9)
浏览 14提问于2020-03-16得票数 1
回答已采纳
我正在使用一家提供实时流的数据供应商的Java API。我想使用Akka streams来处理这个流。 Java API有一个pub sub设计，其工作原理大致如下： Subscription sub = createSubscription();
sub.addListener(new Listener() {
public void eventsReceived(List events) {
for (Event e : events)
buffer.enqueue(e)
}
浏览 32提问于2021-09-28得票数 1
回答已采纳
1回答
嗨，我和Kafka和Akka Streams一起工作。在Kafka中，对于主题MyTestTopic，我有3个分区，数据在高度并发的情况下被推入主题，大致上是1000 QPS，它只会比这个更高。
下面是我为Akka Stream Kafka使用者编写的代码：
final ConsumerSettings<String, byte[]> consumerSettings =
ConsumerSettings.create(kafkaConfig, new StringDeserializer(), new ByteArrayDeserializer())
浏览 5提问于2019-12-09得票数 0
1回答
与像Storm和Flink这样的流处理引擎相比，我什么时候会使用像RX和Project反应堆这样的反应式编程库呢？
我知道这些概念可能不是直接可比的，但我想更好地理解这两个概念之间的关系。
我目前的理解是，两者都关注无界的事件/数据流。
Reactive是一个更低层次的概念，因为它可以实现流处理引擎。我认为Akka Streams的情况就是这样。
流处理引擎的目标似乎是解决横向扩展流处理任务，即聚类等问题。
请澄清这两个概念之间的共同点和差异，每个概念的典型用例以及它们之间的相互关系。
浏览 0提问于2020-03-01得票数 0
我正在寻找一种组件或其他技术来压缩和加密多个大文件(大小超过4 4gb的文件，因此不适合可用于win32进程的内存)为单个文件。我希望加密是非常强大的(256位AES或更好)，但压缩对我来说无关紧要。
现在，我使用的是TJvZlibMultiple组件，它可以创建自己的非Zip文件格式，但我必须创建归档文件，然后在单独的步骤中对其进行加密(我现在使用的是DcpCrypt )。我想在一遍中做同样的事情，使用streams，而不使用与文件大小相等的内存(因此，压缩和加密应该在streams中进行，而不是在内存中)。
我已经看到，并且不想使用任何需要外部dll的东西，比如7zip dll。商业工具，
浏览 3提问于2012-07-27得票数 6
我正在尝试使用akka-streams和akka-http以及alpakka库下载/上传文件到亚马逊S3。我看到两个可能相关的问题..。
我只能下载非常小的文件，最大的一个8kb。
我不能上传更大的文件。它与消息失败。
处理请求时出错：“子流源没有在5000毫秒内物化”。完成500内部服务器错误响应。若要更改默认异常处理行为，请提供自定义ExceptionHandler。akka.stream.impl.SubscriptionTimeoutException:子流源没有在5000毫秒内物化
这是我的路线
pathEnd {
post {
浏览 4提问于2017-09-19得票数 2
我使用Akka应用程序已经有一段时间了。95%的代码是用纯参与者编写的。现在我要把应用程序的一些部分移到Akka Streams。让我了解从Akka流的角度来看，下面的逻辑是怎样的：
+------------+
| CreateUser |
+------------+
|
浏览 0提问于2018-04-08得票数 0
回答已采纳
我一直在跟踪，我遇到了一个错误。我的代码很简单：
object Main extends App {
implicit val system = ActorSystem("QuickStart")
implicit val materializer = ActorMaterializer()
val source: Source[Int, NotUsed] = Source(1 to 100)
source.runForeach(i => println(i)) (materializer)
}
但是，这会导致一个错误：
线程"main“中的异
浏览 4提问于2016-11-02得票数 1
回答已采纳
这是我之前的问题：
我已经成功地使用FileIO.fromPath(Paths.get(file.toURI()在Akka流上发送了文件，它工作得很好。但是，我想在发送文件之前对其进行压缩和加密。我已经创建了方法，打开FileInputStream，通过压缩流路由它，然后通过加密流路由它，现在我想使用Akka流将它定向到套接字中。
文件-> FileInputStream -> CompressedInputStream -> EncryptedInputStream -?> Spring/Akka流
关键是，我可以实时地压缩/加密文件，同时逐块读取它(我没有在磁盘上创
浏览 2提问于2017-02-25得票数 1
回答已采纳
1回答
我正在用akka io/streams编写Java XMPP服务器。根据XMPP规范，连接模式可以在飞行中切换到TLS模式。使用akka.io或akka.streams有什么好的/标准的方法来实现这一点吗？谢谢!
浏览 0提问于2015-12-09得票数 1
我需要做一些类似于这个的事情
我的问题是，我有一个未知的组数，如果mapAsync的并行数少于我在最后一个接收器中得到的组数和错误数。
由于上行误差SynchronousFileSink(/Users/sam/dev/projects/akka-streams/target/log-ERROR.txt) (akka.stream.impl.StreamSubscriptionTimeoutSupport$$anon$2)而拆下(akka.stream.impl.StreamSubscriptionTimeoutSupport$$anon$2)
我试图按照akka 模式指南中的建议，在中
浏览 0提问于2015-08-25得票数 5
回答已采纳
我是akka streams的新手，akka输入了我的代码，与经典的akka一起工作很好，这就是我要做的。
class FlowActor extends Actor{
val log = LoggerFactory.getLogger(this.getClass)
def receive: Receive = {
case word: String =>
val replyTo = word.capitalize
log.info("Actor {}, Capitalizing {}",context.self.path, re
浏览 3提问于2020-06-16得票数 1
回答已采纳
我正在用开发一个流应用程序。我的应用程序如下所示。
flatMap更改键并从单个消息生成多条消息。
使用步骤1中的键使用join和KTable。
transform用于有状态操作。
flatMap将密钥更改为原始密钥，即在步骤1之前。
groupBy具有键，如步骤4，实际上是步骤1。
reduce将记录“合并”到包含JSON数组的单个消息中。
净效果是将传入的消息(以key作为id1)拆分为多个消息(具有不同的键，例如k1、k2等)。使用join和transform增强每条消息。然后，将每条消息的密钥更改为id1。最后，将每个增强消息“合并”为一个消息，其中键为
浏览 10提问于2022-10-17得票数 0
1回答
基于Akka的项目的正确解析器
、、、、
向下拉akka-streams和akka-http-streams添加正确的解析程序是什么？
我添加了
resolvers += "Akka Snapshot Repository" at "http://repo.akka.io/snapshots/"
但Intellij还是不能把神器取下来。
resolvers += "Akka Snapshot Repository" at "http://repo.akka.io/snapshots/"
lazy val root = (project in file(".
浏览 8提问于2018-09-08得票数 0
回答已采纳
在处理各种背压场景时，我实现了这样一种情况:一个订阅者使用缓冲区的速度很慢，而另一个订阅者则消耗掉扔给它的任何东西。使用Scala和Akka Streams。如果愿意，您可以查看代码以及运行它的测试。
我通常尝试开发一个RxJava版本来进行比较，但我在这个版本上卡住了。在Akka Streams中，我可以构建一个图，其中一个源在2个通道上广播，并从这些通道消耗一个慢接收器和一个快速接收器。每个通道可以独立地应用缓冲和节流。在RxJava中，有用于广播的share操作符，但是缓冲和节流逻辑不在Subscriber上，而是在Observable上。因此，我不确定如何应用缓冲和节流，而不影响两个订
浏览 1提问于2016-09-02得票数 2
1回答
了解Akka如何提供背压
、、、、
因此，我们在我们的生产系统中有一个用例，我们可能会使用Akka streams。为了了解Akka streams究竟是如何提供背压的，我想更深入地了解我们的需求。 我们有一个托管部分数据的Solr集群。接下来，我们有一个Play应用程序，它服务于面向客户的前端站点。每个传入的请求最终归结为使用Solr提供的/sql处理程序从Solr获取大量数据。一旦我们从Solr获取整个数据集，我们将其变形后写回Cassandra集群。这可以转换成一个问题，可以使用Akka streams来解决，其中来自/sql处理程序的Solr流将是akka Source，而Cassandra存储将是Sink，而介于两者
浏览 18提问于2020-09-20得票数 0
1回答
现在我正在阅读Play Java的文档。我不明白这些句子解释了apply(RequestHeader header)到底在起什么作用。
方法的返回类型是累加器。蓄能器是阿克卡流水池周围的一层薄薄的一层。一个累加器异步地将元素流累加到一个结果中，它可以通过传入一个Akka流源来运行，这将返回一个CompletionStage，当累加器完成时将被赎回。它与Sink>本质上是一样的，实际上它只不过是这种类型的包装器，但最大的区别是，累加器提供了方便的方法，如map、mapFuture、recover等，用于处理结果，就好像它是一个承诺，Sink要求将所有这些操作包装在mapMateriali
浏览 0提问于2017-01-27得票数 1
akka streams是如何在幕后工作的？
它本质上是一个功拉动()，但可能有多个步骤到功拉动机制？
有人能告诉我源代码是怎么做的吗，或者它是不是太复杂了？:)
浏览 0提问于2017-02-03得票数 0
我正在尝试使用akka streams进行异步http调用。
这就是我尝试过的。
Source(listEndpoints)
.mapAsync(20)(endpoint => Future(Await.result(request(HttpMethods.POST, endpoint, List(authHeader)), timeout)))
.runWith(Sink.seq[HttpResponse])
我在请求方法中使用了akka-http，它返回Future[HttpResponse]
我想我在这里滥用了Future。上面的代码会给我一个Future[
浏览 16提问于2020-11-18得票数 0
回答已采纳
4回答
假设我有一个Java8Streams数组：Stream<T>[] streams，我想创建一个Stream，其中新流的每个元素都是由从每个初始基本流中选择一个元素组成的数组(假设它们都是顺序的)。
例如，如果我有：
streams [ 0 ] returning: ( "A", "B", "C" ),
streams [ 1 ] returning ( "X", "Y", "Z" )
and streams [ 2 ] as ( "0", "1&#
浏览 7提问于2017-07-24得票数 5
回答已采纳
1回答
我试图写一个卡夫卡消费者到websocket流使用反应性- Kafka，akka-http和akka-流。
val publisherActor = actorSystem.actorOf(CommandPublisher.props)
val publisher = ActorPublisher[String](publisherActor)
val commandSource = Source.fromPublisher(publisher) map toMessage
def toMessage(c: String): Message = TextMessage.Stri
浏览 1提问于2016-04-01得票数 6
我正在使用akka创建一个可运行的图形。没有编译时误差，wrt进/出流组件.运行时抛出以下错误：
有什么想法我应该核实一下才能让它运行吗？
requirement failed: The inlets [] and outlets [] must correspond to the inlets [in] and outlets [out]
at scala.Predef$.require(Predef.scala:219)
at akka.stream.Shape.requireSamePortsAs(Shape.scala:168)
at akka.stream.impl.StreamLa
浏览 1提问于2016-02-17得票数 14
回答已采纳
可以分别使用Source.actorPublisher()和Sink.actorSubscriber()方法从参与者创建源和接收器。但是从actor创建Flow是可能的吗？
从概念上讲，似乎没有充分的理由不这样做，因为它同时实现了ActorPublisher和ActorSubscriber特征，但不幸的是，Flow对象没有任何方法来实现这一点。在优秀的博客文章中，它是在Akka Streams的早期版本中完成的，所以问题是在最新的(2.4.9)版本中是否也有可能。
浏览 1提问于2016-08-24得票数 24
回答已采纳
我希望用Flink替换我的基于Akka Streams的流处理器的一部分。目前是否可以使用Akka Streams作为Flink的源，然后在同一代码库中使用Flink作为Akka Streams的源？
Akka Streams的当前流量如下：
// Kafka Source -> Read Flow involving Azure cosmosDB -> Evaluate Flow -> Write Flow -> Logger Flow -> Sink
lazy private val converterGraph: Graph[ClosedShape.t
浏览 2提问于2019-04-24得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
