文档建议反馈控制台
首页
学习
活动
专区
圈层
工具
MCP广场
文章/答案/技术大牛
发布

akka-streams - 从应用角度学习：basic stream parts

实际上很早就写了一系列关于akka-streams的博客。但那个时候纯粹是为了了解akka而去学习的，主要是从了解akka-streams的原理为出发点。...因为akka-streams是akka系列工具的基础，如：akka-http, persistence-query等都是基于akka-streams的，其实没有真正把akka-streams用起来。...不过，现在所处的环境还是逼迫着去真正了解akka-streams的应用场景。现状是这样的：跨入大数据时代，已经有大量的现代IT系统从传统关系数据库转到分布式数据库（非关系数据库）了。...所以流处理应该是分布式数据处理的理想方式了。这是这次写akka-streams的初衷：希望能通过akka-streams来实现分布式数据处理编程。...虽然运算值不能像流元素一样流动，但akka-streams提供了机制让用户选择是否返回某个节点的运算值M。

1.3K10
    • 您找到你想要的搜索结果了吗？
    是的
    没有找到

    面向流的设计思想

    正如面向对象设计以对象为基本设计要素，函数式编程思想以函数为基本设计要素，响应式编程则应该以流为基本设计要素。...这带来设计思想上根本的变化，包括： 以流作为建模的元素 流存在松耦合的上下游关系 以流为重用的单位 对流进行转换、运算、合并与拆分 在Rx框架中，一个流就是一个Observable或者Flowable。...无论哪个流发射了数据，它都会将这两个流最近发射的数据组合起来，并按照指定的函数进行运算。 Akka Stream提出来的Graph更能体现流作为建模元素的思想。...通过这样的可视化图，我们就可以针对这些图中的节点建模为Akka Streams中的Graph Shape。...因此，使用响应式编程，需得围绕“流”为中心进行设计思考，并将其作为一个非常重要的重用元素进行组合。这也就是我所谓的面向流设计（Stream-Oriented Design）的想法来源。

    1.8K30

    PlayScala 2.5.x - 实现完全异步非阻塞的流数据导出

    介绍 从Play2.5.x开始，Play使用Akka Streams实现流处理，废弃了之前的Enumerator/Iteratee Api。...根据官方文档描述，迁移至Akka Streams之后，Play2.5.x的整体性能提升了20%，性能提升相当可观。...该项目目前的流处理功能基于Enumerator/Iteratee实现，Akka Stream的实现放在一个单独的项目开发(RM-AkkaStreams)。...实现 由于ReactiveMongo暂时还没有提供Akka Streams的流处理实现，所以无法直接通过map/flatMap直接返回一个Stream写回响应： @Singleton class TestStreamController...第10行foldBulks方法负责批量从MongoDB数据库读取查询结果，然后以消息形式将数据发送给sourceActor，最后发送一个Status.Success消息表明数据已经发送完毕。

    1.1K40

    sparkstreaming和spark区别

    Spark Streaming 和 Spark 的基本概念Spark StreamingSpark Streaming 是一个流式处理框架，它允许用户以高吞吐量的方式处理实时数据流，Spark Streaming...可以处理来自多种数据源（如 Kafka、Flume、Kinesis 等）的数据，并将连续的数据流拆分成一系列离散的数据批次，这些批次被称为 DStreams（Discretized Streams），...技术教学使用 Spark Streaming要开始使用 Spark Streaming，你需要设置一个 Spark Streaming 上下文，然后从数据源创建 DStreams，定义转换和输出操作，以下是一个简单的示例...，展示了如何使用 Spark Streaming 从一个文本文件源读取数据，并对每个单词进行计数。...("WordCount")val ssc = new StreamingContext(conf, Seconds(1))// 从文本文件源创建 DStreamval lines = ssc.textFileStream

    88910

    通过两个简单的教程来提高你的 awk 技能

    awk 由 Alfred Aho、Peter Weinberger 和 Brian Kernighan（即工具名称中的 A、W 和 K）在 20 世纪 70 年代创建，用于复杂的文本流处理。...它是流编辑器 sed 的配套工具，后者是为逐行处理文本文件而设计的。awk 支持更复杂的结构化程序，是一门完整的编程语言。...awk 如何处理文本流 awk 每次从输入文件或流中一行一行地读取文本，并使用字段分隔符将其解析成若干字段。在 awk 的术语中，当前的缓冲区是一个记录。...第一个任务是设置字段分隔符变量和其他几个脚本需要的变量。你还需要读取并丢弃 CSV 中的第一行，否则会创建一个以 Dear firstname 开头的文件。...一个使用了这个概念的简单的程序就是词频计数器。你可以解析一个文件，在每一行中分解出单词（忽略标点符号），对行中的每个单词进行递增计数器，然后输出文本中出现的前 20 个单词。

    2K20

    提升awk技能的两个教程【译】

    由Alfred Aho, Peter Weinberger, and Brian Kernighan (即awk命名中的A, W, K)于20世纪70年代创建，用于文本流的复杂处理。...awk是流编辑器sed的配套工具，sed是为逐行处理文本文件而设计的。awk则允许更复杂的结构化编程，是一种完整的编程语言。...awk是怎样处理文本流的？ awk从输入文件或流中每次读取一行文本，并使用字段分隔符将其解析为多个字段。awk术语中，当前缓冲区（buffer）是一条记录。...你也需要读取并丢弃proposals.csv的第一行，否则会创建出一个以Dear firstname开头的文件。为了做到这点，需要使用特定的函数getline并在读取之后，把记录计数器重置为0。...一个使用这个概念的简单示例是词频计数器。你可以解析一个文件，提取出每行的单词（忽略标点符号），为该行中的每个单词的计数器递增，然后输出在文本中出现次数在前20的单词。

    5.3K10

    Java8编程思想之Lambda表达式&Stream流式编程&方法引用（method references）

    这些管道通常被组合在一起，在流上形成一条操作管道。 大多数情况下，将对象存储在集合是为了处理他们，因此你将会发现编程焦点从集合转移到了流。流的一个核心好处是，它使得程序更加短小并且更易理解。...我们可以使用 Random 为任意对象集合创建 Supplier。如下是一个文本文件提供字符串对象的例子。...中间操作 中间操作用于从一个流中获取对象，并将对象作为另一个流从后端输出，以连接到其他操作。 跟踪和调试 peek() 操作的目的是帮助调试。它允许你无修改地查看流中的元素。...RandomPair 创建了随机生成的 Pair 对象流。在 Java 中，我们不能直接以某种方式组合两个流。所以这里创建了一个整数流，并且使用 mapToObj() 将其转化成为 Pair 流。...capChars 随机生成的大写字母迭代器从流开始，然后 iterator() 允许我们在 stream() 中使用它。就我所知，这是组合多个流以生成新的对象流的唯一方法。

    2.4K20

    Play For Scala 开发指南 - 第1章 Scala 语言简介

    从2001年开始，Scala经历了如下发展过程： 大事记 2001年诞生于EPFL 2003年发布初始版本 2006年发布2.0版本 2011年Odersky创建Typesafe，后改名Lightbend...Akka包含很多模块，Akka Actor是Akka的核心模块，使用Actor模型实现并发和分布式，可以将你从Java的多线程痛苦中解救出来；Akka Streams可以让你以异步非阻塞的方式处理流数据...，并且支持背压(backpressure)； Akka Http实现了一套基于流的HTTP服务端和客户端；Akka Cluster可以帮你实现一个分布式集群系统；Cluster Sharding可以帮你做集群的分片处理...；Distributed Data可以帮助你在集群之间分享数据；Alpakka可以帮你为Akka Streams集成不同的数据源；Akka Persistence可以帮你处理Actor消息的持久化存储，...Play 刚开始发布的 1.x 版本是基于Java开发， 从 2.0 版本开始，整个框架使用Scala进行了重写。笔者正是从Play 2.0开始，从传统的SSH/SSI转向Play，一直使用至今。

    1.7K60

    反应式架构(1)：基本概念介绍 顶

    有一点需要提醒的是，虽然Java 9已经实现了Reactive Streams，但这并不意味着像RxJava、Reactor、Akka Streams这些流处理框架就没有意义了，事实上恰恰相反。...Reactive Streams的目的在于增强不同框架之间的交互性，提供的是一组最小功能集合，无法满足我们日常的流处理需求，例如组合、过滤、缓存、限流等功能都需要额外实现。...举个例子来说，MongoDB的Java驱动实现了Reactive Streams规范， 开发者使用任何一个流处理框架，仅需要几行代码即可实时监听数据库的变化。...因为是从不同角度描述方法的调用过程，所以这两组概念也可以相互组合，即将线程状态和通知机制进行组合。..., Scala, Kafka and Akka Streams

    2.1K10

    Java8新特性Lambda表达式&Stream流&方法引用最全集锦

    我们可以使用 Random 为任意对象集合创建 Supplier。如下是一个文本文件提供字符串对象的例子。...可使用 flatMap() 解决： 从map返回的每个流都会自动扁平为组成它的字符串。 现在从一个整数流开始，然后使用每个整数去创建更多的随机数。...concat()以参数顺序组合两个流。 如此，我们在每个随机 Integer 流的末尾添加一个 -1 作为标记。你可以看到最终流确实是从一组扁平流中创建的。...RandomPair 创建了随机生成的 Pair 对象流。在 Java 中，我们不能直接以某种方式组合两个流。所以这里创建了一个整数流，并且使用 mapToObj() 将其转化成为 Pair 流。...capChars 随机生成的大写字母迭代器从流开始，然后 iterator() 允许我们在 stream() 中使用它。就我所知，这是组合多个流以生成新的对象流的唯一方法。

    2.7K21

    异步编程 - 14 异步、分布式、基于消息驱动的框架 Akka

    插件和扩展：Akka 提供了丰富的插件和扩展机制，可以轻松集成其他库和框架，如 Akka HTTP、Akka Streams 等，以构建全栈应用程序。...下面看下Akka的特性： 可以更简单地构建并发和分布式系统 Akka基于Actor模型和Streams，让我们可以构建可伸缩的，并且可以高效使用服务器资源，使用多个服务器进行扩展的系统。...内存占用少；每GB堆可以创建约250万个actor（参与者）。 弹性和分散性 分布式系统没有单点故障，具有跨节点的负载平衡和自适应路由。...使用CRDT（Conflict-free Replicated Data Types，无冲突的复制数据类型）实现最终一致性的分布式数据。 反应流数据 具有回压的异步非阻塞流处理。...异步任务执行失败时，任务状态可能丢失，需要引入新的错误信令机制以及从故障中恢复的方法。

    2.3K40

    Java 编程要点之 IO 流详解

    支持所有可能的行结束符，程序可以读取任何广泛使用的操作系统创建的文本文件。 修改 CopyCharacters 来演示如使用面向行的 I/O。...scanning API 使用分隔符模式将其输入分解为标记。formatting API 将数据重新组合成格式良好的，人类可读的形式。...虽然 Scanner 不是流，但你仍然需要关闭它，以表明你与它的底层流执行完成。 调用 useDelimiter() ,指定一个正则表达式可以使用不同的标记分隔符。...Java 支持两种交互方式：标准流（Standard Streams）和通过控制台（Console）。 标准流 标准流是许多操作系统的一项功能。默认情况下，他们从键盘读取输入和写出到显示器。...ObjectStreams 创建与 DataStreams 相同的应用程序。首先，价格现在是 BigDecimal 对象，以更好地代表分数值。

    83320

    Kafka Streams - 抑制

    相反，Kafka Streams是一种优雅的方式，它是一个独立的应用程序。 Kafka Streams应用程序可以用Java/Scala编写。 我的要求是将CDC事件流从多个表中加入，并每天创建统计。...为了做到这一点，我们不得不使用Kafka Streams的抑制功能。 要理解Kafka流的压制概念，我们首先要理解聚合（Aggregation）。...◆聚合的概念 Kafka Streams Aggregation的概念与其他函数式编程（如Scala/Java Spark Streaming、Akka Streams）相当相似。...你可以使用Reduce来组合数值流。上面提到的聚合操作是Reduce的一种通用形式。reduce操作的结果类型不能被改变。在我们的案例中，使用窗口化操作的Reduce就足够了。...为了从压制中刷新聚集的记录，我不得不创建一个虚拟的DB操作（更新任何具有相同内容的表行，如update tableX set id=(select max(id) from tableX);。

    2K10

    Flink实战(五) - DataStream API编程

    最初从各种源（例如，消息队列，套接字流，文件）创建数据流。 结果通过接收器返回，接收器可以例如将数据写入文件或标准输出（例如命令行终端）。...如果要查看大于1的计数，请在5秒内反复键入相同的单词（如果不能快速输入，则将窗口大小从5秒增加☺）。 Socket输入 程序输出 创建一个新数据流，其中包含从套接字无限接收的字符串。...可以从StreamExecutionEnvironment访问几个预定义的流源： 3.1 基于文件 readTextFile(path) TextInputFormat逐行读取文本文件，即符合规范的文件...fromCollection(Iterator, Class) 从迭代器创建数据流。该类指定迭代器返回的数据元的数据类型。 fromElements(T …) 从给定的对象序列创建数据流。...程序可以将多个转换组合成复杂的数据流拓扑。 本节介绍了基本转换，应用这些转换后的有效物理分区以及对Flink 算子链接的见解。

    2.3K10

    Java编程要点之 IO 流详解

    有两种通用的字节到字符的“桥梁”流：InputStreamReader 和 OutputStreamWriter。当没有预包装的字符流类时，使用它们来创建字符流。...支持所有可能的行结束符，程序可以读取任何广泛使用的操作系统创建的文本文件。 修改 CopyCharacters 来演示如使用面向行的 I/O。...scanning API 使用分隔符模式将其输入分解为标记。formatting API 将数据重新组合成格式良好的，人类可读的形式。...Java 支持两种交互方式：标准流（Standard Streams）和通过控制台（Console）。 标准流 标准流是许多操作系统的一项功能。默认情况下，他们从键盘读取输入和写出到显示器。...ObjectStreams 创建与 DataStreams 相同的应用程序。首先，价格现在是 BigDecimal 对象，以更好地代表分数值。

    74510
    点击加载更多

    相关资讯

    热门标签

    更多标签

    活动推荐

      运营活动

      活动名称
      广告关闭

      腾讯云开发者

      扫码关注腾讯云开发者

      扫码关注腾讯云开发者

      领取腾讯云代金券

      热门产品

      热门推荐

      更多推荐

      Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有

      深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 粤公网安备44030502008569号

      腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号

      领券