无处不在,搜你所想

在流处理刚被提出来的时候，很多人认为流处理只能进行做近似的结果或者增量的计算，倘若你想保证其安全性，以 Lamda 架构为基础，利用流处理得到最现在的结果。但同时你需要采用 batch processing 等其他方式来保证其全局的安全性以正确性。

在如此多年的研究结果下，在我看来，流处理并不一定是近似的，或者是仅仅以无法保证真确性为代价而提高速度的一种数据处理方式。相反，流处理应该是一个与全局计算、batch processing 稍微有点不同的计算模型。跟批量处理不同之处在于，批量处理将数据引向计算，而流处理将计算引向数据。这句话大概有点模糊，接下来，我举几个大家熟悉的计算模型例子。

第一个计算模型例子—请求应答模型。

请求应答模型是业务生活中最常用的模型例子。首先提交一个请求到服务方，而服务方可能是一个数据库、也可能是别的存储工具；然后进行等待…等待；最后得到一个回答。这便是一次请求、一次计算、一次回答。该模型非常简单、也极易操作，当你需要延展到多个机器上时，只要简单地增加客户端以及处理器即可成功。但是缺点在于，不能达到大的吞吐量，每提交一次请求，都需要等待时间来获得最终应答的结果。

第二种常见的模型就是批量处理如上图所示。如果请求应答模型在谱系的一端，那么 typo 的另一端则认为是批量处理。当我积累数据数量足够多的时候，一次性提交任务到数据仓库，再进行等待，等待时间短则几秒钟、几分钟，长则几小时，最后才得到最终的结果—所有输入对应的所有输出。该批处理模型的好处在于能够提高其吞吐率，一次的请求和应答可以得出较多结果。但它的缺点是具有高延时性，比如某数据产生时间为上午 6 点钟，用户点击某网页，由于批处理模型，每 12 小时才会运行一次，那么它必须等到上午 6 点到下午 6 点的所有数据完整以后才会进行工作，那么运行结果可能是用户点击的 12 个小时之后。高延迟性是批处理自身带有的特性。

那么什么是流处理呢? 在我看来，流处理就是介于请求应答和批处理之间的一种新型计算模型或者编程模型。流处理并不等待数据的完整性，或者说数据本没有完整性这一讲法，数据本身就是一个数据流，当每个数据流每产生一个新数据的时候立刻被计算出、进行返回，因此数据是源源不断地通向计算，并且源源不断有结果被输出。你可以设想，与等待数据完全完成之后发布到计算上相比，流处理就是将计算移到你数据发生地进行实时计算的方式。

为什么很多人之前有这样一种错觉，他们认为流处理可能存在有丢包的情况、或者说只可以得到近似的结果，其实这是早期的一些数据流处理系统所自带的一些限制。因此以 Lamda 架构为基础，在流处理上需要讨论不同维度的取舍。接下里我将举三个例子，延迟、、成本和正确性。正如很多人之前提及的，在进行流处理时候，其大多数情况需要用时间来换取正确性，或者用更多的成本换取时间等等。

第一个例子，说如果你需要做一个实时的 ETL 处理。而关于 ETL 处理不需要太小的延迟，为达到低成本的一种保证，我们可以忍受几分钟或者 1 分钟的延迟；但是，如果你正在进行一个实时的在线监测，存在着几毫秒的延迟，那么这时候可能更愿意选择花大量的金钱，或者采取一些可能不必要的 possibility 来达到一种低延迟的效果；

第二个例子，假设你在做一个在线付费协议，它也是一个流处理平台。由于在线付费协议可能关乎到其机构，或者其公司的利益所在，因此你会说，我需要保证百分之百的正确性，我不希望有任何丢包情况；

第三个例子，如果你是做一个实时的日志处理，实时收集所有日志，并将其导入 root，在这种情况下，你可能会说，为了降低成本，我愿意付出一小部分正确性的代价，即使不能达到 100%、达到 99.99%、达到 99.9%，这样的结果都可以接受。这本是用户在定义不同流处理应用或者业务的时候应该可以自己做出的选择。但比较遗憾的是，多数早期的流处理平台其实并没有给予用户该种选择，他们自身的设计理念，那就是为了低延迟直接放弃掉正确性，或者说为了更高的吞吐量直接放弃低延迟。

以上是我想分享的关于流处理的一些误会认知，如果我的分享能够让大家带走两个答案的话，我希望这就是一个。我认为流处理仅仅是一种不一样的计算模型或者编程模型，它将计算带到数据上，而不是将数据引用到计算上，并且在流处理的时候，用户往往需要在正确性、延迟性、成本等不同的维度上做出选择。

为什么当我们说到流处理的时候，很多人都在说 Kafka。大多数人在最早接触 Kafka 时会说，Kafka 就是一个分布式发布订阅的消息系统，但是如果我们去观察 Kafka 的最初一些设计特性可发现以下几点内容。第一点，它可以作为一个写在磁盘上的缓存来使用，或者说，并不是仅基于内存来存储流数据，它可以保证数据包不被及时消费时，依然可用且不被丢失；第二点，由于位移的存在提供了逻辑上的顺序，在同一个话题上，第一个数据比第二个数据最先被发布的时候，也可保证在消费时也是永远第一个数据比第二个数据先被消费；第三点，因为 Kafka 是一个公有的大数据中转站，就是说，所有的数据只要在 Kafka 上，永远可以在 Kafka 周围进行业务的开发或者认知事物的开发。接下来我将花费一些时间详细介绍这三点之间的关系。

Kafka 不仅仅是一个订阅消息系统，同时也是一个大规模的流数据平台，那么它提供了什么呢？第一，提供订阅和发布消息；第二，提供一个缓存的流数据存储平台；第三，提供流数据的处理平台。今天，我将着重讨论流式计算在 Kafka 上面的应用。

流式计算在 Kafka 上的应用主要有哪些选项呢？第一个选项就是 DIY，Kafka 提供了两个客户端 —— 一个简单的发布者和一个简单的消费者，我们可以使用这两个客户端进行简单的流处理操作。举个简单的例子，利用消息消费者来实时消费数据，每当得到新的消费数据时，可做一些计算的结果，再通过数据发布者发布到 Kafka 上，或者将它存储到第三方存储系统中。DIY 的流处理需要成本。打个比方，考虑数据的延迟性，考虑不同时间上的管理分配，正如很多人提到的 processing time，这将是我后文会重点提及的概念。以上这些都说明，利用 DIY 做流处理任务、或者做流处理业务的应用都不是非常简单的一件事情。

第二个选项是进行开源、闭源的流处理平台。比如，spark。关于流处理平台的一个公有认知的表示是，如果你想进行流处理操作，首先拿出一个集群，且该集群包含所有必需内容，比如，如果你要用 spark，那么必须用 spark 的 runtime。因为他们划定了你作为一个流处理平台使用者需要用到的所有行为，比如，资源管理系统、参数调配系统、容器配置、代码封装、分发等，以上行为都已被该平台所限定。一旦你选择使用甲就必须用甲套餐装备，如果选择使用乙就必须使用乙套餐装备。有人不禁提出疑问，我能不能既选择流处理平台，又使用自己选择的，我能不能这样做呢？

这个应用场景其实很普遍，举个例子，可异步式微服务处理。什么叫异步式微服务处理？假设 Kafka 作为一个缓存数据，在该缓存区含有很多不同的业务。打个比方，一个网店的机构可以有不同的组、不同的员工，有人负责销售、有人负责商品分发，有人负责价格管理、有人负责在线实时的限流监控，不同的组、不同的员工可能会以不同的时间，或者以不同的代码来更新他们的产品，只要拥有一个异步式缓存机制，即 Kafka，便可扩大该微服务，而不需要他们的任何一个组之间进行同步请求应答机制。

在该微服务情况下，每个小组的喜好、特性并不一致，有的组表示我需要做流处理平台，从 Kafka 读数据，处理完再写回 Kafka，并且想要使用 EWS 把我的应用部署在云端大规模集群上；而另外小组表示我不需要那么复杂，我只是小规模数据，不希望起一个集群，只需起三个机器，并且每个机器有 1GB 内存足以，可进行手动控制操作，不需要资源管理器。那么我们能不能同时满足他们不同的需求呢? 答案就是我接下来要说的第三种选项。

第三种选项是使用一个轻量级流处理的库，而不需要使用一个广泛、复杂的框架或者平台来满足他们不同的需求。在 Kafka 0.10 当中已发布轻量级流处理内容平台，我们可以设想，跟其他客户端发布者和消费者一样，它也是一个客户端，不同之处在于它是一个计算者客户端，一个好用的、功能强大的客户端，并且支持 state processing、Windows 延时的、异步的、甚至不同数据的调控。 最重要的是 Kafka 作为一个库，可以采用多种方法来发布流处理平台的使用。比如，你可以构建一个集群；你可以把它作为一个手提电脑来使用；甚至还可以在黑莓上运行 Kafka。以上都是尤其简单的运行库的概念。

因此我们要做的事情与使用 Kafka 其他的客户端类似，比如发布者、消费者，只要在代码里边加入就可以使用各种各样的 API。当你要调配控制 Kafka Stream 应用的时候，选择最基础的 War File 来运行或者采用 Java、C，甚至资源管理器来运行都是可行的。因为 Kafka Stream 是一个轻量级流处理的库，可支持各种各样的运维方式。

在我们看来，简单的就是美的，只有给用户提供最大的兼容性与最大的延展性，用户才能得到最好的用户体验。

如果接触过 Storm、Spark 等流处理平台的同学可以发现，它们与 Kafka Stream 高阶位 DSL 语言其实有相似之处。如上图所示，首先定义一个 Streams 流， Streams 是从 topic1 中的 topic 获取得到，即定义 Streams、处理 Streams、得到新的 Streams。比如，从 topic1 里面得到两个原始数据流，然后数据流进行 countByKey 得到新的数据流叫做 Counts。那么 counts.to(“topic2”) 是什么意思呢？在获取到新的数据流之后写回 Kafka topic2 内，启动 KafkaStreams 进程，与 Kafka producer、Kafka consumer 类似，让它来运行已定义计算。

正如大家所了解的，API 的使用其实很简单。提供一个简单的 API，用户简单地写入运行逻辑即可运行。但是编程应用总是容易的，而它的复杂程度在于，一旦你开始运维该应用，当你想要把业务拓展到更大规模，或者业务出现变化，或者集群不稳定，需要强大的运维时，运维的程度便显得异常重要，最上面的编程可能只是冰山一角。Kafka Stream 的设计理念是最简单的就是最美的，包括 API、运维、debugging，以及各种各样的方式，都是希望给用户带来最简单的体验。它的核心思想就是把难问题直接给 Kafka 集群本身。

Kafka 的核心思想是什么？就是把这些消息全部存成一个有序日志，所有的消息发布者把消息发布到底端，从某一个逻辑上的位移开始顺序读取所有的消息。它的一个好处在于所有的读和写，尽管都是刷到磁盘上，但都是按照顺序进行，该方式对磁盘的使用比较有效，倘若消费者和发布者隔得比较近，将利用 page cash 直接读数据。

延展性。如上图，提供 topic 以及 topic partitions，即话题与话题分区的机制。每个用户有不同的 topic，每个 topic 可以有多个分区，每个分区可被装载在不同的机器上，当用户提高规模之后，Kafka 只需要简单地增加机器和 topic partitions 数量，或者采用 ROM balance 的方式到不同机器上，即可达到线性延展方式。

以上是 Kafka 最简单的核心思想，接下来我将介绍 Kafka Streams 作为 Kafka 客户端如何利用以上核心思想来设计流处理的平台。数据流其实就是有序的记录或消息，每个消息是一个 Key 加一个 Value，并且 record 与 Kafka 自身 massage 具有一一对应关系。

用户所提供的业务上的计算模型，其实可用拓补结构进行表达。如上图，图的左边。用户首先进行定义数据流，然后对数据流进行计算，得到新的数据流，最终将数据流写回到 Kafka 内。每当用户进行定义的时候，每一步都会变成拓扑结构里面的一个点，每个点通过流进行计算，变成新的流来进行新的连接，最终在 Kafka 内部形成拓扑结构。用户并不需要在意该拓补结构，只需明白定义流、计算流、得到新的流，写回 Kafka。

连接每一个不同的运算单元就是一个 Stream，即 record stream，每一个 Stream 都在源源不断地实时产生 record，每一个 record 是一个 key 加一个 value。利用 Stream Processor 连接 Stream，每个用户定义的流的一个计算单位对应着一个 Stream Processor。

当用户定义每一步计算的时候，就是定义每个拓扑结构里面的每个点，最终把整个拓补结构定义完整到 Kafka Stream 来运行。计算单元其实可分成两个特殊的单元，一个叫做元的计算单元，只有输出流，没有输入流，它们唯一的认同就是从 Kafka 读取数据形成数据流，传递给下方其他数据处理。而 Stream Processor 底端的数据流，没有输出流，只有输入流，它们的功能是把所有输入流写回到 Kafka。Kafka 的运行操作简单，源数据从 Kafka log 读取消息变成数据流，每个消息贯穿整个拓扑结构，最终从 Stream  Processor 写回到 Kafka。以上为 Kafka Stream 运行情况。

用户进行并行发布进程、应用或者多个计算的操作其实也非常简单。Kafka 是一个库，当你用 Kafka 库写成应用，当 record 写入多台机器时，Kafka Stream 库本身就会自动调动 partitions 方式，假设你有两台机器，每台机器上都运行了 Kafka Streams，当它同时进行运行时，不同的 streams application instance 就会从不同的 Kafka partitions 内读取数据来达到并行任务的分发与执行，任务之间没有任何的数据重叠，当你需要更多线性地增长任务时，你只需要在不同的机器上运行同样的 record，所有的 instance 将会自动进行 rebalance，把新的 application 写入，然后获取到延展。

很多人看到不同的计算方式的时候会发现，有的计算方式，比如说 fliter、map，没有“计算状态”需要保存，一个数据进来计算、一个数据出去。但是有的计算，比如说 join、aggregate，就需要动态维护一个“计算状态”，每一次新的信息或者日志进来的时候， Stream 就要进行更新甚至进行读取。后者被称为 Stateful Processing，前者为 Stateless Processing。

那么如何进行管理流处理的 states 呢？有两个通用的方式，一个方式是 remote State，利用远程的数据库或者远程的 key value store 存储所有流处理的 states，每一次计算的时候，发送一个远程请求来读取 states。远程请求的缺点在于需要进行远程的请求和应答。因为 states 存在于 Remove State 上，states 之间可能会有 overlation，不能很好做到 accesstion. 比如我是团队 A，只负责 sell，另外一个是团队 B，只负责 ajustment, 两个不同的流有着不同的 job，但是 state 存在一起，所以两者会相互影响；

另外一个方式是 Local State，意味着所有的 state 和所有的处理单元是并发在一起的，每个单元上存着 state。在 Kafka Stream 里面，每个计算单元之间不需要有任何交互，state 之间亦如此。我们只要把 state 存到 Local 计算单元上就足矣。第一，可以保证 better isolation，它们之间没有任何的 access；第二，local state 可以做到更好的时效性，不需要远程读写。

如上图，在 Kafka 内有 aggregateByKey（…）语句，类似于 Stateful Processing。当用户定义 Stateful Processing 的时候，在 Kafka Stream 库内部就会自动生成 State Strom，且与 aggregate opprate 进行连接，只有该 opprate 能够对该 State Strom 进行读写，因为每个 opprate 有自己独有的 State Strom，可达到 State Strom 完全 Local 化。

当我们有多个并发流处理任务的时候，每个计算单元除了有一个自己的拓扑结构进行计算之外，也有一份 State Store。每个 State Strom 之间是存储完全不相干的流处理信息和数据。

接下来讨论的是 Kafka Streams 里面另一个重要概念，流与数据库表的关系？正如大家所看见的，在 Kafka Streams 内部有两种流—— KStream 与 Ktable，那么什么叫做 KStream？什么叫做 Ktable 呢？在开发 Kafka Streams 时的一个核心出发点是流和它所对应的表或者数据库的 State 彼此之间具有一一影射关系。为什么一一影射呢？

举个例子，假设你有一个上图的数据流，该数据流代表着某张表，即变量的日志或者更新日志。更新日志内含有 Key 和 Valve，比如第三条的更新日志（key1，value3）其实正在更新第 1 日志（key1，value1）的新信息，换句话说，原本 key1 所对应的是 value1，但是在这一时刻被改成对应 value3，如果我们重复更新该日志，我们能够得到什么呢？我们可以得到该表在任意时间段内的一个实时的可视化图。

同理，如果我们只有这样一个表，并且正在不断更新这个表，只要在每次更新时保留该日志，就能够从表反推回该更新日志的数据流所应的所有内容，这就是流和表或者流和状态之间的一一对应关系。总而言之，只要你有一个日志更新流，即可重构回你表状态在任意时间内的 value；如果你有一个表，也可以通过表的更新来找到该表所对应的流。这就是我所说的 A Stream is a changelog of a table ；A table is a materialized view at tiome of a stream. 流和表具有对应关系。

这促使我们定义两种不同的——KStream 和 KTable。KStream 是很普通的数据流，在数据流之间不存在任何因果关系和逻辑关系，可以被认为是 append only Stream。Typo 是更新日志流，每个日志里面相同的 key 所对应的就是对表的更新。那么为什么要定义这两种不同的数据流呢？我举个例子。

如上图，用户购买历史记录。比如 Alice 曾经买过鸡蛋和牛奶，鸡蛋和牛奶这两者之间不存在任何因果关系，Alice 买过牛奶只是在 Alice 买过鸡蛋上很简单的增量。用户雇佣状态的更新日志，比如 Alice 曾经在 LinkedIn 工作，之后信息被更新到 Alice 在微软工作，现在 Alice 在微软工作覆盖了之前的工作信息。

如果以当前的时间状态进行解读这两个流，第一个流显示的信息为 Alice 曾经买过鸡蛋，第二个流信息显示为 Alice 在 LinkedIn 工作。如果将时间往前推，查看更新的数据流信息可以发现，第一个 KStream 显示 Alice 买了鸡蛋又买了牛奶；但是在第二种情况下，Alice 并不是同时在 LinkedIn 和微软工作，而是 Alice 已经在微软工作，不在 LinkedIn 工作了。

为什么两种不同的流有两种定义呢？因为当你做相同操作的时候，比方说简单做一个合计操作，不同的流得出的结果是不一样的。在上者，如果我们将时间往前推，可得出 Alice 的合计结果是 2+3；但是在下面，如果对其进行 KTable 的 aggregate，显示 Alice 的结果是将其原本数值 2 变成 3，而不是 +3 的关系。

在 Kafka Stream 的 DSL 里面有多种不同的 aggregate,reduce 操作等， 不同的数据流可能将 KStream 变成 KTable，也可能把 KTable 变回 KStream，在用户定义如下不同的 operation 的时候，在后台不同状态的流可采用不同计算方式、计算模型。

如上图，KTable。当一条新消息进来时该如何进行拓扑计算呢？举个例子，在该拓扑结构内，Stream2 出现了一个新的 record，即红颜色标记，该标记与第一条 record 颜色相近，因为它们是同个 key，不同 value。Stream2 和 Stream1 进行 join 操作成为一个新的 record，该新 record 会被放入到 KStream joined 里面，然后 KStream joined 进行 aggregate 操作，而 aggregate 操作得到的结果是 state 被更新，新 record 被 append 到 aggregate 流内，但是 append 操作将之前的红颜色 record 复写了，换句话说，因为有了该新 record 的存在，之前红颜色的 record 由于被复写已经不重要了。

如果我们有一个 fault，那么我们如何在 Kafka Stream 上做 fault tolerance？

正如上文所提及的，Tables 和 Stream 之间存在一一影射关系，Kafka Stream 有效地利用了该特性。举个例子，有个 Kafka Stream 的应用业务，该业务有三个并发 task，每个 task 有自己的 local state，每当 State 进行更新时，Kafka Stream 就会自动将更新消息写到更新日志内，更新日志也自动生成。每更新一个状态时，消息日志就被更新该日志上。

比如过了一段时间，中间的 task 坏掉了，那么 Kafka Stream 会做什么呢？首先它会检测异常，自动地在已有的 instance 上重新启动原本坏掉的 task，重新构建 State，那么 State 怎么 build 呢？通过更新 changelog，直到 restore 整个原本正在进行的状态的 restoration，只有新状态被 restore 完整之后才能继续 task 同步计算。

消息回溯也是类似的原理。比方说，某应用已被运行了很多年，发现 stream 流处理计算里面存在 Bug，我们不得不将已计算的结果舍弃，回溯到一个更早的历史时间重新进行计算，即计算回溯。Reprocessing 在 Kafka Stream 也是一种简单的方式，当我们达到某一个位移，比如位移 5，需要进行消息回溯时，用户可以简单地起一个新的状态 -New State，该 State 完全没有任何内容，然后从最早的时间开始重新进行计算，直到计算到赶上现有 task 时候。只需要 switch over 就可以完成消息回溯，且该整个消息回溯过程不需要关闭整个流处理任务。于是很多人便问，那么 Kafka Stream 能不能支持 Streaming processing 呢？

举个例子，我不希望 Kafka Stream 一直在运行，希望它可以每 6 个小时 run 一次，并且每 run 一次可将当前所有已累计的 Kafka massage 全部处理掉。这个操作也很简单，从 outsite  A 开始，一直位移到 B 结束或者到 C 结束，表示已停止整个应用；6 个小时之后当它重启的时候，再从新的位移开始进行下一段的位移，这是批处理计算结果，即从一个 outsite 到另外一个 outsite，紧接着是另外一个 outsite…Kafka Stream 通过位移的控制和管理进行批处理结果，而不需要运行整个 Kafka Stream。

时间管理是流处理上非常重要的观念，同时也是区别于流处理和批量式处理非常重要的概念。很多人都已熟悉 Event Time 和 Processing Time 的区别，Event Time 是每个日志、消息、状态发生的时候所发生的时间，而 Processing Time 是日志被计算和处理的时候所发生的时间。这两者可能并不是完全融合的，可能存在位移，这便是所谓的时间延迟。

如上图，以《星球大战》故事时间和拍摄时间为例。《星球大战》有七步曲，Processing Time 是电影真正拍摄时间，是在现实生活中的时间——1999 年到 2015 年；但是拍摄时间和星球大战所发生时间并不一一对应，存在延迟。对其做流处理时候可以发现，类似 out of order 的现象很常见，比如因为数据量太大而导致数据发生延迟，或者说数据处理发生了延迟等，都会发生延时情况。

那么 Kafka Stream 怎么解决该问题呢？ Kafka Stream 允许给每个日志定义时间戳，该时间戳可以是当前系统时间，也可以是提取时间戳，也可以从当前 record 被生成的时候所提取的时间戳，这些即被定义成 Event Time。类似的，如果 record 是一个 Jason format，将其时间戳提取出来也可被定义成 Event Time。

有如此时间戳，我们可以基于该时间戳进行各式计算，比方说 Windowing 的计算。举个例子，每隔 5 分钟计算一个平均值、总和或者合计，每一个 Windowing 正如上图颜色所示，不同颜色代表不同的时间戳和不同的 Windowing。当你收到一个 record，而该 record 时间戳指向非常未来的时间，你便得到一个非常未来的日志。Kafka 不会直接更新当前的 Windowing，而是会生成该时间戳所对应的 Windowing 更新 aggregate。

同理，倘若你继续计算，你会发现有个古老日志的时间戳指向很早以前的 Windowing。Kafka Stream 可以通过更新原本的 aggregate 来达到这样延时结果。用户在现在时间进行如下定义，比方说定义 Window aggregation，每一个 Windowing 是 5 分钟，但是我希望每个 Windowing 可保持整整一天时间，只要该 Windowing 在当前 24 小时之内依然存在即可做到。

上面已经分享了很多，从ordering到状态，一直到partitioning & scalability ，但是最重要的是所有这些都是由Kafka Stream库自动完成的。我们希望用户不受以上任何一个问题的影响，只是定义自己的业务，上面所有的问题都是由Kafka Stream解决的，虽然只是一个库，但是仍然有足够的能力处理一切。

在Kafka 0.10中发布Kafka流之后，我们将Stream扩展到Java以外的语言，比如Python，或者像SQL这样的高级编程模型，以便用户更容易地定义自己的流处理应用程序。在7月的版本中，我们还将增加一次精确计算的总和。

很多人可能会好奇，Kafka Stream非常好，但是我的数据不是在Kafka中，Kafka Stream只能从Kafka中获取。如何将数据导入Kafka?答案是Kafka Connect，一个简单的数据导入和导出框架。到去年年底，Kafka Connect拥有40种不同规模的连接，从JDBC到HDFS，一直到MYSQL，以及所有可以想到的第三方系统，允许用户轻松地将数据从第三方系统导入和导出到Kafka。

简而言之，回到源头，什么是Kafka?Kafka是一个中央流处理平台，支持消息的发布、消费、传输和存储，以及消息的计算和处理。