Direct messaging from producers to consumers

A number of messaging systems use direct network communication between producers and consumers without going via intermediary nodes:

许多消息系统使用生产者和消费者之间的直接网络通信，而不经过中介节点。

• UDP multicast is widely used in the financial industry for streams such as stock market feeds, where low latency is important [ 8 ]. Although UDP itself is unreliable, application-level protocols can recover lost packets (the producer must remember packets it has sent so that it can retransmit them on demand).

  UDP组播在金融行业广泛应用于像股票市场数据等的流媒体，低延迟很重要。虽然UDP本身不可靠，但应用层协议可以恢复丢失的数据包（生产者必须记得它已经发送的数据包，以便在需要时进行重传）。

• Brokerless messaging libraries such as ZeroMQ [ 9 ] and nanomsg take a similar approach, implementing publish/subscribe messaging over TCP or IP multicast.

  没有经纪人的消息传递库，如ZeroMQ和nanomsg采用类似的方法，通过TCP或IP组播实现发布/订阅消息传递。

• StatsD [ 10 ] and Brubeck [ 7 ] use unreliable UDP messaging for collecting metrics from all machines on the network and monitoring them. (In the StatsD protocol, counter metrics are only correct if all messages are received; using UDP makes the metrics at best approximate [ 11 ]. See also “TCP Versus UDP” .)

  StatsD和Brubeck使用不可靠的UDP消息在网络上从所有机器收集指标并对其进行监控。（在StatsD协议中，只有当收到所有消息时，计数器指标才正确；使用UDP最多只能获得近似的指标[11]。另请参见“TCP与UDP”）

• If the consumer exposes a service on the network, producers can make a direct HTTP or RPC request (see “Dataflow Through Services: REST and RPC” ) to push messages to the consumer. This is the idea behind webhooks [ 12 ], a pattern in which a callback URL of one service is registered with another service, and it makes a request to that URL whenever an event occurs.

  如果消费者在网络上提供了服务，那么生产者可以直接通过HTTP或RPC请求（请参阅“服务数据流：REST和RPC”）向消费者推送消息。这就是Webhook[12]背后的思想，这是一种模式，其中一个服务的回调URL被注册到另一个服务中，并在发生事件时向该URL发出请求。

Although these direct messaging systems work well in the situations for which they are designed, they generally require the application code to be aware of the possibility of message loss. The faults they can tolerate are quite limited: even if the protocols detect and retransmit packets that are lost in the network, they generally assume that producers and consumers are constantly online.

尽管这些直接信息传递系统在它们被设计的情况下工作得很好，但它们通常需要应用程序代码意识到消息丢失的可能性。它们可以容忍的错误相当有限：即使协议检测到并重新传输在网络中丢失的数据包，它们通常假定生产者和消费者始终在线。

If a consumer is offline, it may miss messages that were sent while it is unreachable. Some protocols allow the producer to retry failed message deliveries, but this approach may break down if the producer crashes, losing the buffer of messages that it was supposed to retry.

如果消费者离线，则可能错过在其不可达时发送的消息。一些协议允许生产者重试失败的消息传递，但如果生产者崩溃，失去了应该重试的消息缓冲区，这种方法可能会崩溃。

Message brokers

A widely used alternative is to send messages via a message broker (also known as a message queue ), which is essentially a kind of database that is optimized for handling message streams [ 13 ]. It runs as a server, with producers and consumers connecting to it as clients. Producers write messages to the broker, and consumers receive them by reading them from the broker.

一种广泛使用的替代方式是通过消息代理（也称为消息队列）发送消息，它实际上是一种针对处理消息流进行了优化的数据库[13]。它作为服务器运行，生产者和消费者作为客户端连接到它上面。生产者将消息写入代理，消费者通过从代理中读取消息来接收它们。

By centralizing the data in the broker, these systems can more easily tolerate clients that come and go (connect, disconnect, and crash), and the question of durability is moved to the broker instead. Some message brokers only keep messages in memory, while others (depending on configuration) write them to disk so that they are not lost in case of a broker crash. Faced with slow consumers, they generally allow unbounded queueing (as opposed to dropping messages or backpressure), although this choice may also depend on the configuration.

通过将数据集中存储在代理中，这些系统可以更容易地容忍客户端的连接和断开（连接，断开和崩溃），并且持久性的问题被移至代理。一些消息代理仅将消息保存在内存中，而其他代理（取决于配置）会将它们写入磁盘，以防代理崩溃时消息丢失。面对缓慢的消费者，他们通常允许无限排队（与丢弃消息或反压相对），尽管这种选择也可能取决于配置。

A consequence of queueing is also that consumers are generally asynchronous : when a producer sends a message, it normally only waits for the broker to confirm that it has buffered the message and does not wait for the message to be processed by consumers. The delivery to consumers will happen at some undetermined future point in time—often within a fraction of a second, but sometimes significantly later if there is a queue backlog.

排队的一个后果是消费者通常是异步的：当生产者发送一条消息时，它通常只等待代理确认已缓存该消息，而不等待消息被消费者处理。向消费者的交付将在某个不确定的未来时间发生 - 通常在几分之一秒内，但如果存在队列积压，则有时会更晚。

Message brokers compared to databases

Some message brokers can even participate in two-phase commit protocols using XA or JTA (see “Distributed Transactions in Practice” ). This feature makes them quite similar in nature to databases, although there are still important practical differences between message brokers and databases:

一些消息代理甚至可以使用XA或JTA参与两阶段提交协议（参见“实践中的分布式事务”）。这个功能使它们与数据库非常相似，尽管消息代理和数据库之间仍然存在重要的实际差异：

• Databases usually keep data until it is explicitly deleted, whereas most message brokers automatically delete a message when it has been successfully delivered to its consumers. Such message brokers are not suitable for long-term data storage.

  数据库通常会保留数据直到显式删除，而大多数消息代理在将消息成功交付给其使用者后会自动删除该消息。这样的消息代理不适合长期数据存储。

• Since they quickly delete messages, most message brokers assume that their working set is fairly small—i.e., the queues are short. If the broker needs to buffer a lot of messages because the consumers are slow (perhaps spilling messages to disk if they no longer fit in memory), each individual message takes longer to process, and the overall throughput may degrade [ 6 ].

  由于它们会快速删除消息，大多数消息代理假设它们的工作集相对较小，即队列较短。如果代理需要缓冲大量消息，因为消费者速度较慢（可能将消息溢出到磁盘，如果它们不再适合内存），每个单独的消息处理时间变长，整体吞吐量可能会下降。

• Databases often support secondary indexes and various ways of searching for data, while message brokers often support some way of subscribing to a subset of topics matching some pattern. The mechanisms are different, but both are essentially ways for a client to select the portion of the data that it wants to know about.

  数据库通常支持辅助索引和各种搜索数据的方法，而消息代理通常支持订阅与某些模式匹配的主题子集的一些方式。这些机制不同，但两者本质上都是客户端选择要了解的数据部分的方法。

• When querying a database, the result is typically based on a point-in-time snapshot of the data; if another client subsequently writes something to the database that changes the query result, the first client does not find out that its prior result is now outdated (unless it repeats the query, or polls for changes). By contrast, message brokers do not support arbitrary queries, but they do notify clients when data changes (i.e., when new messages become available).

  查询数据库时，结果通常基于数据的某个时间点的快照；如果另一个客户端稍后向数据库写入了一些更改查询结果的内容，则第一个客户端无法发现其先前的结果已过时（除非它重复查询或轮询更改）。 相比之下，消息代理不支持任意查询，但它们在数据更改时通知客户端（即在新消息变得可用时通知）。

This is the traditional view of message brokers, which is encapsulated in standards like JMS [ 14 ] and AMQP [ 15 ] and implemented in software like RabbitMQ, ActiveMQ, HornetQ, Qpid, TIBCO Enterprise Message Service, IBM MQ, Azure Service Bus, and Google Cloud Pub/Sub [ 16 ].

这是消息代理的传统视角，它被封装在像JMS [14]和AMQP [15]这样的标准中，并在RabbitMQ，ActiveMQ，HornetQ，Qpid，TIBCO企业消息服务，IBM MQ，Azure服务中得到了实现Bus和Google Cloud Pub / Sub [16]。

Multiple consumers

When multiple consumers read messages in the same topic, two main patterns of messaging are used, as illustrated in Figure 11-1 :

当多个消费者在相同的主题中读取消息时，使用了两种主要的消息模式，如图11-1所示：

Load balancing

Each message is delivered to one of the consumers, so the consumers can share the work of processing the messages in the topic. The broker may assign messages to consumers arbitrarily. This pattern is useful when the messages are expensive to process, and so you want to be able to add consumers to parallelize the processing. (In AMQP, you can implement load balancing by having multiple clients consuming from the same queue, and in JMS it is called a shared subscription .)

每条消息都会被发送到其中一个消费者，这样消费者可以共同处理主题中的消息。代理可以任意地将消息分配给消费者。当消息处理成本很高时，这种模式非常有用，因此您希望能够添加消费者来并行处理。 （在AMQP中，您可以通过让多个客户端从同一队列消费来实现负载平衡，在JMS中则称为共享订阅。）

Fan-out

Each message is delivered to all of the consumers. Fan-out allows several independent consumers to each “tune in” to the same broadcast of messages, without affecting each other—the streaming equivalent of having several different batch jobs that read the same input file. (This feature is provided by topic subscriptions in JMS, and exchange bindings in AMQP.)

每条消息都会发送给所有的消费者。扇出允许多个独立的消费者“调谐”到相同的消息广播，而不会相互影响-这就像有几个不同的批处理作业读取相同的输入文件的流媒体一样。(这个特性由JMS中的主题订阅和AMQP中的交换绑定提供。)

Figure 11-1. (a) Load balancing: sharing the work of consuming a topic among consumers; (b) fan-out: delivering each message to multiple consumers.

The two patterns can be combined: for example, two separate groups of consumers may each subscribe to a topic, such that each group collectively receives all messages, but within each group only one of the nodes receives each message.

这两个模式可以结合使用：例如，两个独立的消费者群体可以各自订阅一个主题，以便每个群体共同接收所有消息，但在每个群体中只有一个节点接收每条消息。

Acknowledgments and redelivery

Consumers may crash at any time, so it could happen that a broker delivers a message to a consumer but the consumer never processes it, or only partially processes it before crashing. In order to ensure that the message is not lost, message brokers use acknowledgments : a client must explicitly tell the broker when it has finished processing a message so that the broker can remove it from the queue.

消费者可能随时崩溃，因此经纪人可能会发送消息给消费者，但消费者可能在崩溃之前未能处理完全，或仅部分处理完毕。为确保消息不丢失，消息经纪人使用确认机制：客户端必须明确告诉经纪人已完成对消息的处理，以便经纪人可以将其从队列中移除。

If the connection to a client is closed or times out without the broker receiving an acknowledgment, it assumes that the message was not processed, and therefore it delivers the message again to another consumer. (Note that it could happen that the message actually was fully processed, but the acknowledgment was lost in the network. Handling this case requires an atomic commit protocol, as discussed in “Distributed Transactions in Practice” .)

如果与客户端的连接关闭或超时，而经纪人未收到确认，则假定该消息未被处理，因此将该消息再次传递给另一个消费者。（请注意，实际上该消息可能已完全处理，但在网络中丢失了确认。处理此情况需要原子提交协议，如“实践中的分布式事务”所讨论的。）

When combined with load balancing, this redelivery behavior has an interesting effect on the ordering of messages. In Figure 11-2 , the consumers generally process messages in the order they were sent by producers. However, consumer 2 crashes while processing message m3 , at the same time as consumer 1 is processing message m4 . The unacknowledged message m3 is subsequently redelivered to consumer 1, with the result that consumer 1 processes messages in the order m4 , m3 , m5 . Thus, m3 and m4 are not delivered in the same order as they were sent by producer 1.

当与负载均衡结合使用时，此重新传递行为对消息排序产生有趣的影响。在图11-2中，消费者通常按照制造商发送的顺序处理消息。但是，当消费者2在处理消息m3时崩溃，同时消费者1正在处理消息m4时，未确认的消息m3随后会被重新发送给消费者1，结果是消费者1按照m4，m3，m5的顺序处理消息。因此，m3和m4不按制造商1发送的相同顺序传递。

Figure 11-2. Consumer 2 crashes while processing m3, so it is redelivered to consumer 1 at a later time.

Even if the message broker otherwise tries to preserve the order of messages (as required by both the JMS and AMQP standards), the combination of load balancing with redelivery inevitably leads to messages being reordered. To avoid this issue, you can use a separate queue per consumer (i.e., not use the load balancing feature). Message reordering is not a problem if messages are completely independent of each other, but it can be important if there are causal dependencies between messages, as we shall see later in the chapter.

即使消息代理器尽可能保持消息顺序（符合JMS和AMQP标准的要求），负载均衡与重传结合必然会导致消息被重新排序。为了避免这个问题，可以为每个消费者使用单独的队列（不使用负载均衡功能）。如果消息完全独立，那么消息重新排序不是问题，但是如果消息之间存在因果依赖关系，则可能很重要，这将在本章后面讨论。

Using logs for message storage

A log is simply an append-only sequence of records on disk. We previously discussed logs in the context of log-structured storage engines and write-ahead logs in Chapter 3 , and in the context of replication in Chapter 5 .

一个日志是仅仅在磁盘上追加记录的序列。我们之前在第三章中讨论了日志结构化存储引擎和预写日志，在第五章中讨论了复制的情况下的日志。

The same structure can be used to implement a message broker: a producer sends a message by appending it to the end of the log, and a consumer receives messages by reading the log sequentially. If a consumer reaches the end of the log, it waits for a notification that a new message has been appended. The Unix tool tail -f , which watches a file for data being appended, essentially works like this.

同样的结构可以用来实现一个消息代理：生产者通过将消息附加到日志末尾来发送消息，而消费者通过按顺序读取日志来接收消息。如果消费者到达日志的末尾，它会等待通知新消息已被附加。Unix工具tail -f，它观察文件中是否有数据被附加，基本上就是这样工作的。

In order to scale to higher throughput than a single disk can offer, the log can be partitioned (in the sense of Chapter 6 ). Different partitions can then be hosted on different machines, making each partition a separate log that can be read and written independently from other partitions. A topic can then be defined as a group of partitions that all carry messages of the same type. This approach is illustrated in Figure 11-3 .

为了扩大吞吐量，超过单个磁盘的性能，可以对日志进行分区（如第6章所述）。不同的分区可以托管在不同的机器上，使每个分区成为一个独立的日志，可以独立于其他分区进行读写。然后可以将一个主题定义为一组携带相同类型信息的分区。这种方法在图11-3中说明了。

Within each partition, the broker assigns a monotonically increasing sequence number, or offset , to every message (in Figure 11-3 , the numbers in boxes are message offsets). Such a sequence number makes sense because a partition is append-only, so the messages within a partition are totally ordered. There is no ordering guarantee across different partitions.

在每个分区中，代理人为每个消息分配一个单调递增的序列号或偏移量（在图11-3中，方框中的数字是消息偏移量）。这样的序列号是有意义的，因为分区是仅追加的，所以分区内的消息是完全有序的。跨不同分区没有排序保证。

Figure 11-3. Producers send messages by appending them to a topic-partition file, and consumers read these files sequentially.

Apache Kafka [ 17 , 18 ], Amazon Kinesis Streams [ 19 ], and Twitter’s DistributedLog [ 20 , 21 ] are log-based message brokers that work like this. Google Cloud Pub/Sub is architecturally similar but exposes a JMS-style API rather than a log abstraction [ 16 ]. Even though these message brokers write all messages to disk, they are able to achieve throughput of millions of messages per second by partitioning across multiple machines, and fault tolerance by replicating messages [ 22 , 23 ].

Apache Kafka、Amazon Kinesis Streams 和 Twitter 的 DistributedLog 是基于日志的消息代理，工作方式如下。Google Cloud Pub/Sub 与其架构类似，但提供类似 JMS 的 API 而非日志抽象。尽管这些消息代理将所有消息都写入磁盘，但通过跨多台计算机分区和复制消息，它们能够实现每秒数百万条消息的吞吐量和容错性。

Logs compared to traditional messaging

The log-based approach trivially supports fan-out messaging, because several consumers can independently read the log without affecting each other—reading a message does not delete it from the log. To achieve load balancing across a group of consumers, instead of assigning individual messages to consumer clients, the broker can assign entire partitions to nodes in the consumer group.

日志基于的方法很容易支持扇出式消息传递，因为多个消费者可以独立地读取日志，而不会相互影响——读取一条消息并不会从日志中删除它。为了在多个消费者节点之间实现负载均衡，代理可以将整个分区分配给该消费者组中的节点，而不是将单个消息分配给消费者客户端。

Each client then consumes all the messages in the partitions it has been assigned. Typically, when a consumer has been assigned a log partition, it reads the messages in the partition sequentially, in a straightforward single-threaded manner. This coarse-grained load balancing approach has some downsides:

每个客户端都会消费分配给它的分区中的所有消息。通常，当消费者分配到一个日志分区后，会按顺序逐个读取分区中的消息，采用简单的单线程方式。这种粗粒度的负载均衡方法存在一些缺点：

• The number of nodes sharing the work of consuming a topic can be at most the number of log partitions in that topic, because messages within the same partition are delivered to the same node. ⁱ

  消费一个主题的节点数量最多只能是该主题的日志分区数，因为同一分区内的消息会被传递到相同的节点。

• If a single message is slow to process, it holds up the processing of subsequent messages in that partition (a form of head-of-line blocking; see “Describing Performance” ).

  如果一个单独的消息处理较慢，它会阻塞该分区后续消息的处理（一种先到先处理的形式；参见“描述性能”）。

Thus, in situations where messages may be expensive to process and you want to parallelize processing on a message-by-message basis, and where message ordering is not so important, the JMS/AMQP style of message broker is preferable. On the other hand, in situations with high message throughput, where each message is fast to process and where message ordering is important, the log-based approach works very well.

因此，在处理信息可能很昂贵且您想要按消息进行并行处理的情况下，而且消息排序并不那么重要时，首选JMS/AMQP样式的消息代理。另一方面，在高消息吞吐量的情况下，每个消息都快速处理且消息排序很重要时，基于日志的方法非常有效。

Consumer offsets

Consuming a partition sequentially makes it easy to tell which messages have been processed: all messages with an offset less than a consumer’s current offset have already been processed, and all messages with a greater offset have not yet been seen. Thus, the broker does not need to track acknowledgments for every single message—it only needs to periodically record the consumer offsets. The reduced bookkeeping overhead and the opportunities for batching and pipelining in this approach help increase the throughput of log-based systems.

按顺序消耗分区可以轻松确定已处理过的消息：具有比消费者当前偏移量小的偏移量的所有消息已经处理完毕，具有更大偏移量的所有消息尚未被查看。因此，代理不需要跟踪每个单个消息的确认 - 它只需要定期记录消费者偏移量。这种方法中减少的簿记开销以及管道化和批处理的机会有助于提高基于日志的系统的吞吐量。

This offset is in fact very similar to the log sequence number that is commonly found in single-leader database replication, and which we discussed in “Setting Up New Followers” . In database replication, the log sequence number allows a follower to reconnect to a leader after it has become disconnected, and resume replication without skipping any writes. Exactly the same principle is used here: the message broker behaves like a leader database, and the consumer like a follower.

这个偏移实际上非常类似于单主数据库复制中常见的日志序列号，我们在“设置新的跟随者”中讨论过。在数据库复制中，日志序列号允许跟随者在与主服务器断开连接后重新连接到主服务器，并在不跳过任何写入的情况下恢复复制。这里使用完全相同的原则：消息代理行为就像一个主数据库，而消费者则像一个跟随者。

If a consumer node fails, another node in the consumer group is assigned the failed consumer’s partitions, and it starts consuming messages at the last recorded offset. If the consumer had processed subsequent messages but not yet recorded their offset, those messages will be processed a second time upon restart. We will discuss ways of dealing with this issue later in the chapter.

如果一个消费者节点失败，消费者组中的另一个节点将被分配到失败的消费者分区，并从最后记录的偏移量开始消费消息。如果消费者已经处理了后续消息但尚未记录它们的偏移量，则在重新启动后这些消息将被再次处理。我们将在本章后面讨论处理这个问题的方法。

Disk space usage

If you only ever append to the log, you will eventually run out of disk space. To reclaim disk space, the log is actually divided into segments, and from time to time old segments are deleted or moved to archive storage. (We’ll discuss a more sophisticated way of freeing disk space later.)

如果你只是向日志追加数据，最终你会耗尽磁盘空间。为了回收磁盘空间，日志实际上被分成了段，而旧的段会定期删除或移动到归档存储中。（我们稍后将讨论一种更复杂的释放磁盘空间的方法。）

This means that if a slow consumer cannot keep up with the rate of messages, and it falls so far behind that its consumer offset points to a deleted segment, it will miss some of the messages. Effectively, the log implements a bounded-size buffer that discards old messages when it gets full, also known as a circular buffer or ring buffer . However, since that buffer is on disk, it can be quite large.

这意味着如果一个慢消费者无法跟上消息的速度，它会落后很远，导致其消费者偏移指向已删除的段，这会导致它错过一些消息。实际上，该记录实现了一个有界大小的缓冲区，当它变满时会丢弃旧的消息，也被称为环形缓冲区或环形缓存。但是，由于该缓冲区在磁盘上，它可以非常大。

Let’s do a back-of-the-envelope calculation. At the time of writing, a typical large hard drive has a capacity of 6 TB and a sequential write throughput of 150 MB/s. If you are writing messages at the fastest possible rate, it takes about 11 hours to fill the drive. Thus, the disk can buffer 11 hours’ worth of messages, after which it will start overwriting old messages. This ratio remains the same, even if you use many hard drives and machines. In practice, deployments rarely use the full write bandwidth of the disk, so the log can typically keep a buffer of several days’ or even weeks’ worth of messages.

让我们进行草稿纸计算。在撰写本文时，一台典型的大型硬盘容量为6 TB，顺序写入吞吐量为150 MB/s。如果您以最快速度写入消息，填充驱动器需要约11小时。因此，磁盘可以缓存11小时的消息，之后它将开始覆盖旧的消息。即使使用多个硬盘和机器，这个比率仍然保持不变。实际上，在部署中很少使用磁盘的完整写入带宽，因此日志通常可以保持数天或甚至数周的消息缓冲。

Regardless of how long you retain messages, the throughput of a log remains more or less constant, since every message is written to disk anyway [ 18 ]. This behavior is in contrast to messaging systems that keep messages in memory by default and only write them to disk if the queue grows too large: such systems are fast when queues are short and become much slower when they start writing to disk, so the throughput depends on the amount of history retained.

日志的吞吐量始终保持相对恒定，无论消息保留多长时间，因为无论如何每个消息都会被写入磁盘。这种行为与默认将消息保留在内存中，仅在队列过大时才将其写入磁盘的消息系统形成对比。这些系统在队列较短时速度很快，当它们开始向磁盘写入时会变得慢得多，因此吞吐量取决于保留的历史数量。

When consumers cannot keep up with producers

At the beginning of “Messaging Systems” we discussed three choices of what to do if a consumer cannot keep up with the rate at which producers are sending messages: dropping messages, buffering, or applying backpressure. In this taxonomy, the log-based approach is a form of buffering with a large but fixed-size buffer (limited by the available disk space).

在“消息系统”的开头，我们讨论了三种选择，如果消费者无法跟上生产者发送消息的速度：丢弃消息、缓冲或施加反压力。在这种分类中，基于日志的方法是一种具有大但固定大小缓冲区（由可用磁盘空间限制）的缓冲。

If a consumer falls so far behind that the messages it requires are older than what is retained on disk, it will not be able to read those messages—so the broker effectively drops old messages that go back further than the size of the buffer can accommodate. You can monitor how far a consumer is behind the head of the log, and raise an alert if it falls behind significantly. As the buffer is large, there is enough time for a human operator to fix the slow consumer and allow it to catch up before it starts missing messages.

如果消费者落后太远，需要的消息比磁盘上保留的消息还要旧，那么它将无法读取这些消息。因此，经纪人有效地删除更早的消息，这些消息回溯的时间超过缓冲区大小。您可以监视消费者落后于日志头的程度，并在其显著落后时发出警报。由于缓冲区很大，因此人工操作员有足够的时间修复缓慢的消费者并使其赶上，以免错过消息。

Even if a consumer does fall too far behind and starts missing messages, only that consumer is affected; it does not disrupt the service for other consumers. This fact is a big operational advantage: you can experimentally consume a production log for development, testing, or debugging purposes, without having to worry much about disrupting production services. When a consumer is shut down or crashes, it stops consuming resources—the only thing that remains is its consumer offset.

即使消费者落后太多开始错过信息，也只影响到该消费者本身，不会影响其他消费者的服务。这个事实是一个极大的运营优势：你可以实验性地消费生产日志，用于开发、测试或调试目的，而不必担心会干扰生产服务。当消费者关闭或崩溃时，它停止消耗资源，唯一留下的是消费者偏移量。

This behavior also contrasts with traditional message brokers, where you need to be careful to delete any queues whose consumers have been shut down—otherwise they continue unnecessarily accumulating messages and taking away memory from consumers that are still active.

此行为与传统的消息代理相比也存在明显的对比，传统的消息代理需要谨慎删除已经关闭的消费者所使用的队列，否则这些队列会持续积累消息，占用消费者还需使用的内存空间。

Replaying old messages

We noted previously that with AMQP- and JMS-style message brokers, processing and acknowledging messages is a destructive operation, since it causes the messages to be deleted on the broker. On the other hand, in a log-based message broker, consuming messages is more like reading from a file: it is a read-only operation that does not change the log.

之前我们注意到，使用AMQP和JMS风格的消息代理时，处理和确认消息是一个破坏性的操作，因为它会导致消息在代理中被删除。另一方面，在基于日志的消息代理中，消费消息更像是从文件中读取：它是一个只读操作，不会改变日志。

The only side effect of processing, besides any output of the consumer, is that the consumer offset moves forward. But the offset is under the consumer’s control, so it can easily be manipulated if necessary: for example, you can start a copy of a consumer with yesterday’s offsets and write the output to a different location, in order to reprocess the last day’s worth of messages. You can repeat this any number of times, varying the processing code.

数据处理的唯一副作用，除了任何消费者的输出之外，就是消费者偏移量向前移动。但偏移量在消费者的控制之下，因此如果需要，很容易进行操作：例如，您可以使用昨天的偏移量启动消费者的副本，并将输出写入不同的位置，以重新处理最后一天的消息。您可以重复这个过程任意次数，可以变换处理代码。

This aspect makes log-based messaging more like the batch processes of the last chapter, where derived data is clearly separated from input data through a repeatable transformation process. It allows more experimentation and easier recovery from errors and bugs, making it a good tool for integrating dataflows within an organization [ 24 ].

这个方面使基于日志的消息更像上一章的批处理，通过可重复的转换过程清楚地将派生数据与输入数据分离。它允许更多的实验和更容易地从错误和漏洞中恢复，使其成为组织内集成数据流的好工具[24]。 这方面使基于日志的消息更像前面章节中的批处理模式，通过可重复转换过程将衍生数据与输入数据明确区别开来。这使得数据集成更加灵活，容易恢复出错和解决漏洞，也是组织内数据流集成的重要工具。[24]

Implementing change data capture

We can call the log consumers derived data systems , as discussed in the introduction to Part III : the data stored in the search index and the data warehouse is just another view onto the data in the system of record. Change data capture is a mechanism for ensuring that all changes made to the system of record are also reflected in the derived data systems so that the derived systems have an accurate copy of the data.

我们可以将日志消耗者派生的数据系统称为第三部分引言中讨论的系统，搜索索引和数据仓库存储的数据仅是系统记录中数据的另一种视图。更改数据捕获是一种机制，可确保所有对系统记录所做的更改也反映在派生数据系统中，以便派生系统具有准确的数据副本。

Essentially, change data capture makes one database the leader (the one from which the changes are captured), and turns the others into followers. A log-based message broker is well suited for transporting the change events from the source database, since it preserves the ordering of messages (avoiding the reordering issue of Figure 11-2 ).

本质上，更改数据捕获将一个数据库设为领导者（从中捕获更改），并将其他数据库转变为追随者。基于日志的消息代理非常适合传输来自源数据库的更改事件，因为它保留了消息的顺序（避免了图11-2中的重新排序问题）。

Database triggers can be used to implement change data capture (see “Trigger-based replication” ) by registering triggers that observe all changes to data tables and add corresponding entries to a changelog table. However, they tend to be fragile and have significant performance overheads. Parsing the replication log can be a more robust approach, although it also comes with challenges, such as handling schema changes.

数据库触发器可以用于实现更改数据捕获（参见“基于触发器的复制”），通过注册观察所有数据表更改并将对应条目添加到更改日志表的触发器。然而，它们往往很容易出错并且具有显着的性能开销。解析复制日志可以是更强大的方法，但它也带来了挑战，如处理模式更改。

LinkedIn’s Databus [ 25 ], Facebook’s Wormhole [ 26 ], and Yahoo!’s Sherpa [ 27 ] use this idea at large scale. Bottled Water implements CDC for PostgreSQL using an API that decodes the write-ahead log [ 28 ], Maxwell and Debezium do something similar for MySQL by parsing the binlog [ 29 , 30 , 31 ], Mongoriver reads the MongoDB oplog [ 32 , 33 ], and GoldenGate provides similar facilities for Oracle [ 34 , 35 ].

LinkedIn的Databus[25]、Facebook的Wormhole[26]和Yahoo！的Sherpa[27]大规模地使用了这个想法。Bottled Water使用API解码预写日志[28]为PostgreSQL实现CDC，Maxwell和Debezium通过解析binlog为MySQL做了类似的事情[29,30,31]，Mongoriver读取MongoDB oplog[32,33]，GoldenGate为Oracle提供类似的设施[34,35]。

Like message brokers, change data capture is usually asynchronous: the system of record database does not wait for the change to be applied to consumers before committing it. This design has the operational advantage that adding a slow consumer does not affect the system of record too much, but it has the downside that all the issues of replication lag apply (see “Problems with Replication Lag” ).

像消息代理一样，更改数据捕获通常是异步的：记录数据库的系统在提交更改之前不会等待消费者应用更改。该设计具有运营优势，即添加一个缓慢的消费者不会对系统的记录产生太大影响，但它的副作用是所有复制延迟的问题都适用（请参见“复制延迟的问题”）。

Initial snapshot

If you have the log of all changes that were ever made to a database, you can reconstruct the entire state of the database by replaying the log. However, in many cases, keeping all changes forever would require too much disk space, and replaying it would take too long, so the log needs to be truncated.

如果您拥有数据库中所有更改的日志记录，可以通过重播日志来重建数据库的整个状态。然而，在许多情况下，永久保留所有更改将需要太多的磁盘空间，并且重放将花费太长时间，因此需要截断日志。

Building a new full-text index, for example, requires a full copy of the entire database—it is not sufficient to only apply a log of recent changes, since it would be missing items that were not recently updated. Thus, if you don’t have the entire log history, you need to start with a consistent snapshot, as previously discussed in “Setting Up New Followers” .

建立新的全文索引，例如，需要完整复制整个数据库-仅应用最近更改的日志是不够的，因为它会缺少未最近更新的项目。因此，如果您没有完整的日志历史记录，您需要从一致的快照开始，如“设置新的关注者”中所讨论的那样。

The snapshot of the database must correspond to a known position or offset in the change log, so that you know at which point to start applying changes after the snapshot has been processed. Some CDC tools integrate this snapshot facility, while others leave it as a manual operation.

数据库的快照必须对应于更改日志中的已知位置或偏移量，这样在快照处理后，您就知道从哪个点开始应用更改。一些CDC工具会集成这种快照功能，而其他工具则将其留作手动操作。

Log compaction

If you can only keep a limited amount of log history, you need to go through the snapshot process every time you want to add a new derived data system. However, log compaction provides a good alternative.

如果你只能保留有限数量的日志历史记录，每次想添加新的派生数据系统时就需要进行快照过程。然而，日志压缩提供了一个很好的替代方案。

We discussed log compaction previously in “Hash Indexes” , in the context of log-structured storage engines (see Figure 3-2 for an example). The principle is simple: the storage engine periodically looks for log records with the same key, throws away any duplicates, and keeps only the most recent update for each key. This compaction and merging process runs in the background.

我们之前在“哈希索引”一章中讨论了日志压缩，其中的上下文是基于日志结构的存储引擎（参见图3-2作为示例）。原理很简单：存储引擎周期性地查找具有相同键的日志记录，丢弃任何副本，并仅保留每个键的最新更新。这个压缩和合并的过程在后台运行。

In a log-structured storage engine, an update with a special null value (a tombstone ) indicates that a key was deleted, and causes it to be removed during log compaction. But as long as a key is not overwritten or deleted, it stays in the log forever. The disk space required for such a compacted log depends only on the current contents of the database, not the number of writes that have ever occurred in the database. If the same key is frequently overwritten, previous values will eventually be garbage-collected, and only the latest value will be retained.

在基于日志结构的存储引擎中，通过使用特定的空值（墓碑），更新表示密钥已被删除，并在日志压缩期间将其删除。但只要密钥未被覆盖或删除，它就会永远保留在日志中。这样压缩日志所需的磁盘空间仅取决于数据库的当前内容，不取决于数据库中以前执行的写入次数。如果经常覆盖同一密钥，则先前的值最终将被垃圾回收，仅保留最新的值。

The same idea works in the context of log-based message brokers and change data capture. If the CDC system is set up such that every change has a primary key, and every update for a key replaces the previous value for that key, then it’s sufficient to keep just the most recent write for a particular key.

在基于日志的消息代理和变更数据捕获的上下文中，同样的想法适用。如果CDC系统设置为每个更改都有一个主键，并且每个关键字的更新替换该关键字的先前值，则仅保留特定关键字的最新写入即足够。

Now, whenever you want to rebuild a derived data system such as a search index, you can start a new consumer from offset 0 of the log-compacted topic, and sequentially scan over all messages in the log. The log is guaranteed to contain the most recent value for every key in the database (and maybe some older values)—in other words, you can use it to obtain a full copy of the database contents without having to take another snapshot of the CDC source database.

现在，每当您想重新构建派生数据系统，例如搜索索引，您可以从日志压缩主题的偏移0开始启动新的消费者，并按顺序扫描日志中的所有消息。日志保证包含数据库中每个键的最新值（可能还有一些旧值），换句话说，您可以使用它获取数据库内容的完整副本，而无需对CDC源数据库进行另一个快照。

This log compaction feature is supported by Apache Kafka. As we shall see later in this chapter, it allows the message broker to be used for durable storage, not just for transient messaging.

这个日志压缩功能由Apache Kafka支持。正如我们在本章后面所将看到的，这使得消息代理可以被用作持久化存储，而不仅仅是短暂的消息传递。

API support for change streams

Increasingly, databases are beginning to support change streams as a first-class interface, rather than the typical retrofitted and reverse-engineered CDC efforts. For example, RethinkDB allows queries to subscribe to notifications when the results of a query change [ 36 ], Firebase [ 37 ] and CouchDB [ 38 ] provide data synchronization based on a change feed that is also made available to applications, and Meteor uses the MongoDB oplog to subscribe to data changes and update the user interface [ 39 ].

越来越多的数据库开始支持变更流作为一流的接口，而不是典型的后续和逆向工程的CDC努力。例如，RethinkDB允许查询订阅通知查询结果更改（36），Firebase（37）和CouchDB（38）基于变更流提供数据同步，也可供应用程序使用，Meteor使用MongoDB oplog来订阅数据更改并更新用户界面（39）。

VoltDB allows transactions to continuously export data from a database in the form of a stream [ 40 ]. The database represents an output stream in the relational data model as a table into which transactions can insert tuples, but which cannot be queried. The stream then consists of the log of tuples that committed transactions have written to this special table, in the order they were committed. External consumers can asynchronously consume this log and use it to update derived data systems.

VoltDB允许事务连续以流的形式从数据库中导出数据。该数据库将输出流在关系数据模型中表示为一个表，事务可以插入元组，但不能查询。然后，该流由已提交的事务写入到此特殊表中的元组日志组成，按照它们提交的顺序。外部消费者可以异步消费此日志并使用它来更新衍生数据系统。

Kafka Connect [ 41 ] is an effort to integrate change data capture tools for a wide range of database systems with Kafka. Once the stream of change events is in Kafka, it can be used to update derived data systems such as search indexes, and also feed into stream processing systems as discussed later in this chapter.

Kafka Connect是将变更数据捕捉工具整合到各种数据库系统和Kafka的一种努力。一旦变更事件的流在Kafka中，它就可以用来更新派生数据系统，例如搜索索引，同时也可以馈送到流处理系统，正如本章后面所讨论的那样。

Deriving current state from the event log

An event log by itself is not very useful, because users generally expect to see the current state of a system, not the history of modifications. For example, on a shopping website, users expect to be able to see the current contents of their cart, not an append-only list of all the changes they have ever made to their cart.

事件日志本身并不是很有用，因为用户通常希望看到系统的当前状态，而不是修改历史。例如，在购物网站上，用户希望能够看到其购物车的当前内容，而不是一个仅包含其购物车所有更改的追加列表。

Thus, applications that use event sourcing need to take the log of events (representing the data written to the system) and transform it into application state that is suitable for showing to a user (the way in which data is read from the system [ 47 ]). This transformation can use arbitrary logic, but it should be deterministic so that you can run it again and derive the same application state from the event log.

因此，使用事件溯源的应用程序需要记录事件日志（代表写入系统的数据），并将其转化为适合向用户展示的应用程序状态（系统中读取数据的方式[47]）。这种转化可以使用任何逻辑，但应该是确定性的，这样您可以再次运行它，并从事件日志中派生相同的应用程序状态。

Like with change data capture, replaying the event log allows you to reconstruct the current state of the system. However, log compaction needs to be handled differently:

像更改数据捕获一样，重放事件日志允许您重建系统的当前状态。然而，需要以不同的方式处理日志压缩：

• A CDC event for the update of a record typically contains the entire new version of the record, so the current value for a primary key is entirely determined by the most recent event for that primary key, and log compaction can discard previous events for the same key.

  CDC事件用于更新记录时，通常包含完整的新记录版本，因此主键的当前值完全取决于该主键的最新事件，并且日志压缩可以丢弃相同键的先前事件。

• On the other hand, with event sourcing, events are modeled at a higher level: an event typically expresses the intent of a user action, not the mechanics of the state update that occurred as a result of the action. In this case, later events typically do not override prior events, and so you need the full history of events to reconstruct the final state. Log compaction is not possible in the same way.

  另一方面，事件溯源模型更高层次地建模事件：事件通常表达用户操作的意图，而不是由于该操作导致的状态更新的机制。在这种情况下，后续事件通常不会覆盖先前事件，因此您需要完整的事件历史记录来重建最终状态。日志压缩无法以相同的方式进行。

Applications that use event sourcing typically have some mechanism for storing snapshots of the current state that is derived from the log of events, so they don’t need to repeatedly reprocess the full log. However, this is only a performance optimization to speed up reads and recovery from crashes; the intention is that the system is able to store all raw events forever and reprocess the full event log whenever required. We discuss this assumption in “Limitations of immutability” .

使用事件源应用程序通常具有某种机制来存储当前状态的快照，该快照来自事件日志，以便它们不需要重复处理完整的日志。但是，这只是一种性能优化，可加快读取和从崩溃中恢复的速度；意图是系统能够永久存储所有原始事件并在需要时重新处理完整的事件日志。我们在“不可变性的局限性”中讨论了这种假设。

Commands and events

The event sourcing philosophy is careful to distinguish between events and commands [ 48 ]. When a request from a user first arrives, it is initially a command: at this point it may still fail, for example because some integrity condition is violated. The application must first validate that it can execute the command. If the validation is successful and the command is accepted, it becomes an event, which is durable and immutable.

事件溯源哲学特别注重区分事件和命令[48]。当用户的请求首次到达时，它最初是一个命令：此时它仍可能失败，例如因为某些完整性条件被破坏。应用程序必须先验证它是否能执行此命令。如果验证成功且命令被接受，则成为一个持久且不可变的事件。

For example, if a user tries to register a particular username, or reserve a seat on an airplane or in a theater, then the application needs to check that the username or seat is not already taken. (We previously discussed this example in “Fault-Tolerant Consensus” .) When that check has succeeded, the application can generate an event to indicate that a particular username was registered by a particular user ID, or that a particular seat has been reserved for a particular customer.

例如，如果用户尝试注册特定的用户名，或者预留飞机或剧院的座位，那么应用程序需要检查该用户名或座位是否已经被占用。（我们之前在“容错共识”中讨论过这个例子。）当检查成功时，应用程序可以生成一个事件，指示特定的用户名已经被特定的用户ID注册，或者特定的座位已经被特定的客户预订。

At the point when the event is generated, it becomes a fact . Even if the customer later decides to change or cancel the reservation, the fact remains true that they formerly held a reservation for a particular seat, and the change or cancellation is a separate event that is added later.

当事件被触发时，它便成为了一个事实。即使顾客后来决定改变或取消预订，这个事实仍然成立，即他们曾经持有预订特定座位的凭证，并且更改或取消是一个后来被添加的单独事件。

A consumer of the event stream is not allowed to reject an event: by the time the consumer sees the event, it is already an immutable part of the log, and it may have already been seen by other consumers. Thus, any validation of a command needs to happen synchronously, before it becomes an event—for example, by using a serializable transaction that atomically validates the command and publishes the event.

事件流的消费者不允许拒绝事件：当消费者看到事件时，它已经是日志不可变的一部分，而且可能已经被其他消费者看到了。因此，任何命令的验证都需要同步发生，即在成为事件之前，通过使用序列化事务原子验证命令并发布事件。

Alternatively, the user request to reserve a seat could be split into two events: first a tentative reservation, and then a separate confirmation event once the reservation has been validated (as discussed in “Implementing linearizable storage using total order broadcast” ). This split allows the validation to take place in an asynchronous process.

Alternatively可以翻译成“或者”，用户请求预留座位可以分成两个事件：首先是暂时预留，然后是在预留得到确认后进行的确认事件（如“使用总序列广播实现线性化存储”中所讨论的）。这种分割允许验证过程在异步处理中进行。

Advantages of immutable events

Immutability in databases is an old idea. For example, accountants have been using immutability for centuries in financial bookkeeping. When a transaction occurs, it is recorded in an append-only ledger , which is essentially a log of events describing money, goods, or services that have changed hands. The accounts, such as profit and loss or the balance sheet, are derived from the transactions in the ledger by adding them up [ 53 ].

数据库中的不可变性是一个古老的概念。例如，会计师在财务簿记中已经使用不可变性几个世纪了。当发生交易时，它被记录在一个附加日志中，该日志本质上是描述已经转移的资金、商品或服务事件的日志。损益表或资产负债表等帐户是通过将其加起来从日志中派生出来的交易[53]。

If a mistake is made, accountants don’t erase or change the incorrect transaction in the ledger—instead, they add another transaction that compensates for the mistake, for example refunding an incorrect charge. The incorrect transaction still remains in the ledger forever, because it might be important for auditing reasons. If incorrect figures, derived from the incorrect ledger, have already been published, then the figures for the next accounting period include a correction. This process is entirely normal in accounting [ 54 ].

如果出现错误，会计师不会在账簿中擦除或更改不正确的交易——相反，他们会添加另一笔交易来补偿错误，例如退还错误的费用。不正确的交易仍然会永久保存在账簿中，因为这可能对审核原因很重要。如果已经发布了从错误账簿派生的不正确数字，则下一个会计期间的数字将包括校正。这个过程在会计中是非常正常的 [54]。

Although such auditability is particularly important in financial systems, it is also beneficial for many other systems that are not subject to such strict regulation. As discussed in “Philosophy of batch process outputs” , if you accidentally deploy buggy code that writes bad data to a database, recovery is much harder if the code is able to destructively overwrite data. With an append-only log of immutable events, it is much easier to diagnose what happened and recover from the problem.

尽管审计能力在金融系统中尤为重要，但对许多其他系统也有益，这些系统并不受到如此严格的监管。如“批处理输出哲学”中所讨论的，如果您不小心部署了有错误的代码，将糟糕的数据写入数据库，如果该代码能够具有破坏性地覆盖数据，则恢复将更加困难。使用不可变时间轴的追加日志记录事件，诊断问题和恢复变得更加容易。

Immutable events also capture more information than just the current state. For example, on a shopping website, a customer may add an item to their cart and then remove it again. Although the second event cancels out the first event from the point of view of order fulfillment, it may be useful to know for analytics purposes that the customer was considering a particular item but then decided against it. Perhaps they will choose to buy it in the future, or perhaps they found a substitute. This information is recorded in an event log, but would be lost in a database that deletes items when they are removed from the cart [ 42 ].

不可变事件还可以捕捉比当前状态更多的信息。例如，在购物网站上，客户可能会将商品添加到购物车中，然后再将其移除。尽管第二个事件从订单履行的角度来看会取消第一个事件，但为了分析目的，知道客户正在考虑某个商品，但最终决定不购买，可能很有用。也许他们将来会选择购买它，或者可能找到替代品。这些信息记录在事件日志中，但在从购物车中删除商品的数据库中将丢失。

Deriving several views from the same event log

Moreover, by separating mutable state from the immutable event log, you can derive several different read-oriented representations from the same log of events. This works just like having multiple consumers of a stream ( Figure 11-5 ): for example, the analytic database Druid ingests directly from Kafka using this approach [ 55 ], Pistachio is a distributed key-value store that uses Kafka as a commit log [ 56 ], and Kafka Connect sinks can export data from Kafka to various different databases and indexes [ 41 ]. It would make sense for many other storage and indexing systems, such as search servers, to similarly take their input from a distributed log (see “Keeping Systems in Sync” ).

此外，通过将可变状态与不可变事件日志分开，您可以从相同的事件日志中获得多种不同的读取导向表示。这就像拥有多个流的消费者一样（图11-5）：例如，分析数据库Druid使用此方法直接从Kafka摄取[55]，Pistachio是一种分布式键值存储，它使用Kafka作为提交日志[56]，而Kafka Connect Sinks可以将数据从Kafka导出到各种不同的数据库和索引[41]。许多其他存储和索引系统，例如搜索服务器，同样可以从分布式日志中获取它们的输入（请参阅“保持系统同步”）。

Having an explicit translation step from an event log to a database makes it easier to evolve your application over time: if you want to introduce a new feature that presents your existing data in some new way, you can use the event log to build a separate read-optimized view for the new feature, and run it alongside the existing systems without having to modify them. Running old and new systems side by side is often easier than performing a complicated schema migration in an existing system. Once the old system is no longer needed, you can simply shut it down and reclaim its resources [ 47 , 57 ].

将事件日志明确地翻译到数据库中，有助于随着时间的推移，演变应用程序：如果您想引入一种新功能，以某种新方式呈现现有数据，您可以使用事件日志为新功能构建单独的读取优化视图，在不修改它们的情况下与现有系统并行运行。在现有系统中执行复杂的模式迁移比在旧系统旁边运行旧和新系统通常更容易。一旦不再需要旧系统，您可以简单地关闭它并回收其资源[47、57]。

Storing data is normally quite straightforward if you don’t have to worry about how it is going to be queried and accessed; many of the complexities of schema design, indexing, and storage engines are the result of wanting to support certain query and access patterns (see Chapter 3 ). For this reason, you gain a lot of flexibility by separating the form in which data is written from the form it is read, and by allowing several different read views. This idea is sometimes known as command query responsibility segregation (CQRS) [ 42 , 58 , 59 ].

如果您不必担心数据如何查询和访问，那么存储数据通常相当简单；模式设计、索引和存储引擎的许多复杂性都是为了支持特定的查询和访问模式（请参见第3章）。由于这个原因，通过将数据写入和读取的方式分离，以及允许多个不同的读取视图，可以获得很多的灵活性。这个想法有时被称为命令查询责任分离（CQRS）[42，58，59]。

The traditional approach to database and schema design is based on the fallacy that data must be written in the same form as it will be queried. Debates about normalization and denormalization (see “Many-to-One and Many-to-Many Relationships” ) become largely irrelevant if you can translate data from a write-optimized event log to read-optimized application state: it is entirely reasonable to denormalize data in the read-optimized views, as the translation process gives you a mechanism for keeping it consistent with the event log.

传统的数据库和模式设计方法是基于一种谬论，即数据必须以查询的形式写入。如果您可以将数据从写优化的事件日志翻译成读优化的应用程序状态，则有关规范化和去规范化的辩论（请参见“多对一和多对多关系”）将变得相对无关紧要：在读优化视图中去规范化数据是完全合理的，因为翻译过程为您提供了一种保持一致性的机制。

In “Describing Load” we discussed Twitter’s home timelines, a cache of recently written tweets by the people a particular user is following (like a mailbox). This is another example of read-optimized state: home timelines are highly denormalized, since your tweets are duplicated in all of the timelines of the people following you. However, the fan-out service keeps this duplicated state in sync with new tweets and new following relationships, which keeps the duplication manageable.

在“描述载荷”一节中，我们讨论了Twitter的主页时间线，这是一个由特定用户正在关注的人最近撰写的推文的缓存（类似于邮箱）。这是另一个读取优化状态的例子：主页时间线高度非规范化，因为你的推文在所有关注你的人的时间线中都会重复。然而，分扇服务使这个重复的状态与新的推文和新的关注关系同步，从而使重复变得可管理。

Concurrency control

The biggest downside of event sourcing and change data capture is that the consumers of the event log are usually asynchronous, so there is a possibility that a user may make a write to the log, then read from a log-derived view and find that their write has not yet been reflected in the read view. We discussed this problem and potential solutions previously in “Reading Your Own Writes” .

事件溯源和变更数据捕获的最大缺点是事件日志的消费者通常是异步的，因此存在一个可能性，即用户可能会将其写入日志，然后从日志派生的视图中读取，并发现其写入尚未在读取视图中反映出来。我们在“读取自己的写入”中曾经讨论过这个问题和潜在的解决方案。

One solution would be to perform the updates of the read view synchronously with appending the event to the log. This requires a transaction to combine the writes into an atomic unit, so either you need to keep the event log and the read view in the same storage system, or you need a distributed transaction across the different systems. Alternatively, you could use the approach discussed in “Implementing linearizable storage using total order broadcast” .

一个解决方案是在将事件追加到日志时同步执行读视图的更新。这需要一个事务将写操作组合成一个原子单元，因此你需要将事件日志和读视图保存在同一个存储系统中，或者你需要在不同系统之间使用分布式事务。另外，你可以使用“使用总序广播实现可线性化存储”的方法。

On the other hand, deriving the current state from an event log also simplifies some aspects of concurrency control. Much of the need for multi-object transactions (see “Single-Object and Multi-Object Operations” ) stems from a single user action requiring data to be changed in several different places. With event sourcing, you can design an event such that it is a self-contained description of a user action. The user action then requires only a single write in one place—namely appending the events to the log—which is easy to make atomic.

另一方面，从事件日志中获取当前状态也简化了并发控制的某些方面。大部分需要多对象事务（参见“单对象和多对象操作”）的原因在于单个用户操作需要在多个不同位置更改数据。使用事件源，您可以设计一个事件，使其成为用户操作的独立描述。然后，用户操作仅需要在一个地方进行单个写入 - 即将事件附加到日志中 - 这很容易使其原子化。

If the event log and the application state are partitioned in the same way (for example, processing an event for a customer in partition 3 only requires updating partition 3 of the application state), then a straightforward single-threaded log consumer needs no concurrency control for writes—by construction, it only processes a single event at a time (see also “Actual Serial Execution” ). The log removes the nondeterminism of concurrency by defining a serial order of events in a partition [ 24 ]. If an event touches multiple state partitions, a bit more work is required, which we will discuss in Chapter 12 .

如果事件日志和应用状态以相同的方式进行分区（例如，仅需更新应用状态的分区3以处理客户的事件），则简单的单线程日志消费者不需要并发控制来进行写操作——由于构造原因，它一次只处理一个事件（另请参阅“实际串行执行”）。日志通过定义分区中事件的序列顺序来消除并发的非确定性[24]。如果事件涉及多个状态分区，则需要更多工作，在第12章中我们将讨论这些。

Limitations of immutability

Many systems that don’t use an event-sourced model nevertheless rely on immutability: various databases internally use immutable data structures or multi-version data to support point-in-time snapshots (see “Indexes and snapshot isolation” ). Version control systems such as Git, Mercurial, and Fossil also rely on immutable data to preserve version history of files.

许多不使用事件溯源模型的系统仍然依赖于不变性：各种数据库内部使用不可变数据结构或多版本数据来支持时间点快照（请参见“索引和快照隔离”）。版本控制系统如Git、Mercurial和Fossil也依赖于不变数据来保留文件版本历史记录。

To what extent is it feasible to keep an immutable history of all changes forever? The answer depends on the amount of churn in the dataset. Some workloads mostly add data and rarely update or delete; they are easy to make immutable. Other workloads have a high rate of updates and deletes on a comparatively small dataset; in these cases, the immutable history may grow prohibitively large, fragmentation may become an issue, and the performance of compaction and garbage collection becomes crucial for operational robustness [ 60 , 61 ].

保留所有更改的不可变历史记录是否可行取决于数据集中的变化量。某些工作负载主要添加数据，很少更新或删除；它们易于变为不可变。其他工作负载在相对较小的数据集上具有高更新和删除率；在这些情况下，不可变历史记录可能会增长到无法承受的大小，碎片化可能会成为一个问题，压缩和垃圾收集的性能变得至关重要 [60、61]。

Besides the performance reasons, there may also be circumstances in which you need data to be deleted for administrative reasons, in spite of all immutability. For example, privacy regulations may require deleting a user’s personal information after they close their account, data protection legislation may require erroneous information to be removed, or an accidental leak of sensitive information may need to be contained.

除了性能原因，由于管理原因，您可能需要删除数据，尽管存在不可变性。例如，隐私规定可能要求在用户关闭账户后删除其个人信息，数据保护法规可能要求删除错误信息，或者敏感信息的意外泄露可能需要被控制。

In these circumstances, it’s not sufficient to just append another event to the log to indicate that the prior data should be considered deleted—you actually want to rewrite history and pretend that the data was never written in the first place. For example, Datomic calls this feature excision [ 62 ], and the Fossil version control system has a similar concept called shunning [ 63 ].

在这种情况下，仅仅将另一个事件附加到日志中以指示应将先前的数据视为已删除是不足够的 - 你实际上想要重写历史并假装第一次没有写入数据。例如，Datomic称此功能为切除[62]，而Fossil版本控制系统具有类似的概念称为shunning[63]。

Truly deleting data is surprisingly hard [ 64 ], since copies can live in many places: for example, storage engines, filesystems, and SSDs often write to a new location rather than overwriting in place [ 52 ], and backups are often deliberately immutable to prevent accidental deletion or corruption. Deletion is more a matter of “making it harder to retrieve the data” than actually “making it impossible to retrieve the data.” Nevertheless, you sometimes have to try, as we shall see in “Legislation and self-regulation” .

真正删除数据的难度出人意料 [64]，因为副本可以存在于许多地方: 例如，存储引擎、文件系统和SSD通常在新位置写入而不是原地覆盖 [52]，备份通常故意不可变，以防止意外删除或损坏。删除更多地是“让检索数据变得更难”而不是真正“使检索数据变得不可能”。虽然如此，有时您必须尝试，正如我们将在“立法和自我调节”中看到的那样。

Complex event processing

Complex event processing (CEP) is an approach developed in the 1990s for analyzing event streams, especially geared toward the kind of application that requires searching for certain event patterns [ 65 , 66 ]. Similarly to the way that a regular expression allows you to search for certain patterns of characters in a string, CEP allows you to specify rules to search for certain patterns of events in a stream.

复杂事件处理（CEP）是开发于1990年代的一种方法，用于分析事件流，特别适用于需要搜索特定事件模式的应用程序[65，66]。类似于正则表达式允许您在字符串中搜索特定模式的字符的方式，CEP允许您指定规则搜索事件流中的某些事件模式。

CEP systems often use a high-level declarative query language like SQL, or a graphical user interface, to describe the patterns of events that should be detected. These queries are submitted to a processing engine that consumes the input streams and internally maintains a state machine that performs the required matching. When a match is found, the engine emits a complex event (hence the name) with the details of the event pattern that was detected [ 67 ].

CEP系统通常使用高级声明性查询语言（如SQL）或图形用户界面来描述应检测到的事件模式。这些查询将提交给处理引擎，该引擎消耗输入流并内部维护状态机以执行所需的匹配。当找到匹配时，引擎发出一个带有被检测到的事件模式细节的复杂事件（因此得名）。

In these systems, the relationship between queries and data is reversed compared to normal databases. Usually, a database stores data persistently and treats queries as transient: when a query comes in, the database searches for data matching the query, and then forgets about the query when it has finished. CEP engines reverse these roles: queries are stored long-term, and events from the input streams continuously flow past them in search of a query that matches an event pattern [ 68 ].

在这些系统中，与普通数据库相比，查询和数据之间的关系被颠倒了。通常，数据库会持久地存储数据，并将查询视为短暂的：当有查询时，数据库将搜索与查询匹配的数据，然后在完成后就忘记了查询。CEP引擎颠倒了这些角色：查询被长期存储，而来自输入流的事件不断流过它们，以寻找与事件模式匹配的查询[68]。

Implementations of CEP include Esper [ 69 ], IBM InfoSphere Streams [ 70 ], Apama, TIBCO StreamBase, and SQLstream. Distributed stream processors like Samza are also gaining SQL support for declarative queries on streams [ 71 ].

CEP 的实现包括 Esper [69]，IBM InfoSphere Streams [70]，Apama，TIBCO StreamBase 和 SQLstream。分布式流处理器像 Samza 也在逐渐获得 SQL 对流查询的支持 [71]。

Stream analytics

Another area in which stream processing is used is for analytics on streams. The boundary between CEP and stream analytics is blurry, but as a general rule, analytics tends to be less interested in finding specific event sequences and is more oriented toward aggregations and statistical metrics over a large number of events—for example:

流处理使用的另一个领域是流分析。 CEP 和流分析之间的界限模糊不清，但一般规则是分析倾向于不关心特定事件序列，而更倾向于对大量事件进行聚合和统计度量。例如：

• Measuring the rate of some type of event (how often it occurs per time interval)

  测量某种类型事件的频率（每个时间间隔发生的频率）

• Calculating the rolling average of a value over some time period

  计算某一时期内某个值的滚动平均值。

• Comparing current statistics to previous time intervals (e.g., to detect trends or to alert on metrics that are unusually high or low compared to the same time last week)

  将当前统计数据与以前的时间间隔进行比较（例如，以便检测趋势或警报指标是否与上周同期明显偏高或偏低）。

Such statistics are usually computed over fixed time intervals—for example, you might want to know the average number of queries per second to a service over the last 5 minutes, and their 99th percentile response time during that period. Averaging over a few minutes smoothes out irrelevant fluctuations from one second to the next, while still giving you a timely picture of any changes in traffic pattern. The time interval over which you aggregate is known as a window , and we will look into windowing in more detail in “Reasoning About Time” .

这类统计通常是在固定的时间间隔内计算的——例如，您可能想知道过去5分钟内服务的平均每秒查询次数，以及在这段时间内它们的99th百分位响应时间。在几分钟内平均化可以平滑不相关的波动，同时仍能及时地给您展示任何流量模式的变化。您聚合的时间间隔称为窗口，在“时间推理”中详细探讨窗口。

Stream analytics systems sometimes use probabilistic algorithms, such as Bloom filters (which we encountered in “Performance optimizations” ) for set membership, HyperLogLog [ 72 ] for cardinality estimation, and various percentile estimation algorithms (see “Percentiles in Practice” ). Probabilistic algorithms produce approximate results, but have the advantage of requiring significantly less memory in the stream processor than exact algorithms. This use of approximation algorithms sometimes leads people to believe that stream processing systems are always lossy and inexact, but that is wrong: there is nothing inherently approximate about stream processing, and probabilistic algorithms are merely an optimization [ 73 ].

流分析系统有时使用概率算法，比如布隆过滤器（我们在“性能优化”中遇到）用于集合成员、HyperLogLog用于基数估计以及各种百分位估计算法（详见“百分位数实践”）。概率算法会产生近似结果，但其优点是需要远比精确算法更少的内存资源。这种运用近似算法有时会让人们认为流处理系统总是有损和不精确的，但这是错误的：流处理本质上并非近似，而概率算法只是一种优化。

Many open source distributed stream processing frameworks are designed with analytics in mind: for example, Apache Storm, Spark Streaming, Flink, Concord, Samza, and Kafka Streams [ 74 ]. Hosted services include Google Cloud Dataflow and Azure Stream Analytics.

许多开源分布式流处理框架都是以分析为重点设计的，例如Apache Storm、Spark Streaming、Flink、Concord、Samza和Kafka Streams [74]。托管服务包括Google Cloud Dataflow和Azure Stream Analytics。

Maintaining materialized views

We saw in “Databases and Streams” that a stream of changes to a database can be used to keep derived data systems, such as caches, search indexes, and data warehouses, up to date with a source database. We can regard these examples as specific cases of maintaining materialized views (see “Aggregation: Data Cubes and Materialized Views” ): deriving an alternative view onto some dataset so that you can query it efficiently, and updating that view whenever the underlying data changes [ 50 ].

在“数据库和流”中，我们看到可以使用数据库的更改流来保持派生数据系统（例如缓存、搜索索引和数据仓库）与源数据库保持最新。我们可以将这些示例视为维护物化视图的特定案例（请参见“聚合：数据立方体和物化视图”）：导出一些数据集的替代视图，以便您可以高效地查询它，并在基础数据更改时更新该视图[50]。

Similarly, in event sourcing, application state is maintained by applying a log of events; here the application state is also a kind of materialized view. Unlike stream analytics scenarios, it is usually not sufficient to consider only events within some time window: building the materialized view potentially requires all events over an arbitrary time period, apart from any obsolete events that may be discarded by log compaction (see “Log compaction” ). In effect, you need a window that stretches all the way back to the beginning of time.

在事件溯源中，通过应用一系列事件日志来维护应用状态；这里应用状态也是某种物化视图。与流分析场景不同的是，仅考虑某个时间窗口内的事件通常是不足够的：构建物化视图可能需要在任意时间段内获取所有事件，除了可能被日志压缩（参见“日志压缩”）丢弃的过时事件。实际上，您需要一个一直延伸到时间开始的窗口。

In principle, any stream processor could be used for materialized view maintenance, although the need to maintain events forever runs counter to the assumptions of some analytics-oriented frameworks that mostly operate on windows of a limited duration. Samza and Kafka Streams support this kind of usage, building upon Kafka’s support for log compaction [ 75 ].

原则上，任何流处理器都可以用于实时视图维护，尽管需要永久维护事件与一些面向分析的框架的假设相矛盾，这些框架主要基于有限持续时间的窗口操作。Samza和Kafka Streams支持这种使用方式，基于Kafka的日志压缩支持 [75]。

Search on streams

Besides CEP, which allows searching for patterns consisting of multiple events, there is also sometimes a need to search for individual events based on complex criteria, such as full-text search queries.

除了CEP，它允许搜索由多个事件组成的模式，有时还需要根据复杂条件搜索个别事件，例如全文搜索查询。

For example, media monitoring services subscribe to feeds of news articles and broadcasts from media outlets, and search for any news mentioning companies, products, or topics of interest. This is done by formulating a search query in advance, and then continually matching the stream of news items against this query. Similar features exist on some websites: for example, users of real estate websites can ask to be notified when a new property matching their search criteria appears on the market. The percolator feature of Elasticsearch [ 76 ] is one option for implementing this kind of stream search.

例如，媒体监控服务订阅来自媒体机构的新闻文章和广播，搜索任何涉及感兴趣的公司、产品或主题的新闻。这是通过预先制定搜索查询，然后不断将新闻流与此查询进行匹配来完成的。一些网站上存在类似的功能：例如，房地产网站的用户可以要求在市场上出现符合其搜索标准的新房产时收到通知。 Elasticsearch 的 percolator 功能是实现这种流搜索的选项之一。

Conventional search engines first index the documents and then run queries over the index. By contrast, searching a stream turns the processing on its head: the queries are stored, and the documents run past the queries, like in CEP. In the simplest case, you can test every document against every query, although this can get slow if you have a large number of queries. To optimize the process, it is possible to index the queries as well as the documents, and thus narrow down the set of queries that may match [ 77 ].

传统搜索引擎首先索引文档，然后在索引上运行查询。相比之下，搜索流程将处理颠倒过来：将查询存储，然后文档通过查询运行，就像在CEP中一样。在最简单的情况下，可以对每个文档进行每个查询的测试，但如果有大量查询，则可能速度较慢。为优化这一过程，可以索引查询以及文档，从而缩小可能匹配的查询集合[77]。

Message passing and RPC

In “Message-Passing Dataflow” we discussed message-passing systems as an alternative to RPC—i.e., as a mechanism for services to communicate, as used for example in the actor model. Although these systems are also based on messages and events, we normally don’t think of them as stream processors:

在“消息传递数据流”中，我们讨论了消息传递系统作为RPC的一种替代方案——即作为服务之间通信的机制，例如在actor模型中使用。虽然这些系统也基于消息和事件，但我们通常不将它们视为流处理器：

• Actor frameworks are primarily a mechanism for managing concurrency and distributed execution of communicating modules, whereas stream processing is primarily a data management technique.

  演员框架主要是用于管理并发和分布式执行通信模块的机制，而流处理主要是一种数据管理技术。

• Communication between actors is often ephemeral and one-to-one, whereas event logs are durable and multi-subscriber.

  演员之间的沟通往往是短暂的一对一交流，而事件日志是持久的、可多次订阅的。

• Actors can communicate in arbitrary ways (including cyclic request/response patterns), but stream processors are usually set up in acyclic pipelines where every stream is the output of one particular job, and derived from a well-defined set of input streams.

  演员可以用任意方式进行沟通(包括循环的请求/响应模式),但流处理器通常设定为非循环的管道,其中每个流都是一个特定工作的输出,并来源于一个明确定义的输入流集合.

That said, there is some crossover area between RPC-like systems and stream processing. For example, Apache Storm has a feature called distributed RPC , which allows user queries to be farmed out to a set of nodes that also process event streams; these queries are then interleaved with events from the input streams, and results can be aggregated and sent back to the user [ 78 ]. (See also “Multi-partition data processing” .)

RPC式系统和流处理之间存在一些交叉区域。例如，Apache Storm具有名为“分布式RPC”的功能，允许将用户查询分配到一组同时处理事件流的节点；然后，这些查询将与输入流中的事件交错，并可聚合结果并发送回用户。请参见“多分区数据处理”。

It is also possible to process streams using actor frameworks. However, many such frameworks do not guarantee message delivery in the case of crashes, so the processing is not fault-tolerant unless you implement additional retry logic.

使用 actor 框架也可以处理数据流。然而，许多这样的框架在发生崩溃时无法保证消息传递，所以除非您实现了额外的重试逻辑，否则处理不具备容错能力。

Event time versus processing time

There are many reasons why processing may be delayed: queueing, network faults (see “Unreliable Networks” ), a performance issue leading to contention in the message broker or processor, a restart of the stream consumer, or reprocessing of past events (see “Replaying old messages” ) while recovering from a fault or after fixing a bug in the code.

处理可能被延迟的原因有很多：排队，网络故障（参见“不可靠网络”），性能问题导致消息代理或处理器中的争用，流消费者的重启，或者在修复代码中的错误后从故障中恢复或重新处理过去的事件（参见“重新播放旧消息”）。

Moreover, message delays can also lead to unpredictable ordering of messages. For example, say a user first makes one web request (which is handled by web server A), and then a second request (which is handled by server B). A and B emit events describing the requests they handled, but B’s event reaches the message broker before A’s event does. Now stream processors will first see the B event and then the A event, even though they actually occurred in the opposite order.

此外，消息延迟还会导致消息的顺序变得不可预测。例如，一个用户首先请求一个网页（由服务器A处理），然后请求第二个网页（服务器B处理）。A和B都会发布关于它们处理的请求的事件，但是B的事件先到达消息代理，然后才是A的事件。现在，流处理器会先看到B的事件，然后再看到A的事件，尽管它们实际上是相反的顺序发生的。

If it helps to have an analogy, consider the Star Wars movies: Episode IV was released in 1977, Episode V in 1980, and Episode VI in 1983, followed by Episodes I, II, and III in 1999, 2002, and 2005, respectively, and Episode VII in 2015 [ 80 ]. ⁱⁱ If you watched the movies in the order they came out, the order in which you processed the movies is inconsistent with the order of their narrative. (The episode number is like the event timestamp, and the date when you watched the movie is the processing time.) As humans, we are able to cope with such discontinuities, but stream processing algorithms need to be specifically written to accommodate such timing and ordering issues.

如果需要一个类比的话，可以考虑《星球大战》电影系列：第四集在1977年上映，第五集在1980年，第六集在1983年上映，接着是第一集、第二集和第三集，在1999年、2002年和2005年上映，最后是第七集在2015年上映[80]。如果您按照它们上映的顺序观看电影，那么您处理电影的顺序与它们叙述的顺序是不一致的。(剧集编号就像事件时间戳，您观看电影的日期就是处理时间。)作为人类，我们能够应对这种不连贯性，但流处理算法需要专门编写以适应这种时间和顺序问题。

Confusing event time and processing time leads to bad data. For example, say you have a stream processor that measures the rate of requests (counting the number of requests per second). If you redeploy the stream processor, it may be shut down for a minute and process the backlog of events when it comes back up. If you measure the rate based on the processing time, it will look as if there was a sudden anomalous spike of requests while processing the backlog, when in fact the real rate of requests was steady ( Figure 11-7 ).

混淆事件时间和处理时间会导致错误的数据。例如，假设你有一个流处理器，用于测量请求的速率（每秒请求的数量）。如果你重新部署流处理器，那么它可能会被关闭一分钟，并在重新启动后处理事件的积压。如果你根据处理时间来测量速率，则会看起来似乎请求的突然异常峰值，但实际上请求的实际速率是稳定的（图11-7）。

Figure 11-7. Windowing by processing time introduces artifacts due to variations in processing rate.

Knowing when you’re ready

A tricky problem when defining windows in terms of event time is that you can never be sure when you have received all of the events for a particular window, or whether there are some events still to come.

定义事件时间窗口时的一个棘手问题在于无法确定何时已接收到某个窗口的所有事件，或者是否还有一些事件需要到来。

For example, say you’re grouping events into one-minute windows so that you can count the number of requests per minute. You have counted some number of events with timestamps that fall in the 37th minute of the hour, and time has moved on; now most of the incoming events fall within the 38th and 39th minutes of the hour. When do you declare that you have finished the window for the 37th minute, and output its counter value?

例如，假设您正在将事件分组到每分钟窗口中，以便可以计算每分钟的请求数量。您已经计算出一些时间戳位于小时的第37分钟内的事件数量，但时间已经过去了。现在，大部分传入事件都在小时的第38和39分钟内。您应该在何时声明已完成第37分钟的窗口，并输出其计数器值？

You can time out and declare a window ready after you have not seen any new events for a while, but it could still happen that some events were buffered on another machine somewhere, delayed due to a network interruption. You need to be able to handle such straggler events that arrive after the window has already been declared complete. Broadly, you have two options [ 1 ]:

在一段时间内没有观察到任何新事件，可以超时且声明窗口已准备好，但仍可能存在某些事件在其他机器上被缓存或由于网络中断而延迟。您需要能够处理在窗口已经被声明为完成后到达的这种拖延事件。一般而言，您有两个选择[1]:

1. Ignore the straggler events, as they are probably a small percentage of events in normal circumstances. You can track the number of dropped events as a metric, and alert if you start dropping a significant amount of data.

   忽略那些迷路的事件，因为它们在正常情况下可能只占事件的一小部分。你可以把丢失的事件数作为指标进行跟踪，并在你开始丢失大量数据时发出警报。

2. Publish a correction , an updated value for the window with stragglers included. You may also need to retract the previous output.

   发布一份更正的通知，包括剩余者在内的更新数值。您可能需要撤回以前的输出。

In some cases it is possible to use a special message to indicate, “From now on there will be no more messages with a timestamp earlier than t ,” which can be used by consumers to trigger windows [ 81 ]. However, if several producers on different machines are generating events, each with their own minimum timestamp thresholds, the consumers need to keep track of each producer individually. Adding and removing producers is trickier in this case.

有些情况下，可以使用特殊信息来表示“从现在开始，将不再有比t更早的时间戳的信息”，消费者可以使用这些信息来触发窗口[81]。然而，如果有多个位于不同机器上生成事件的生产者，每个生产者都有自己的最小时间戳阈值，消费者需要单独跟踪每个生产者。在这种情况下，添加和删除生产者更加棘手。

Whose clock are you using, anyway?

Assigning timestamps to events is even more difficult when events can be buffered at several points in the system. For example, consider a mobile app that reports events for usage metrics to a server. The app may be used while the device is offline, in which case it will buffer events locally on the device and send them to a server when an internet connection is next available (which may be hours or even days later). To any consumers of this stream, the events will appear as extremely delayed stragglers.

在事件可以被缓冲在系统的几个点之后分配时间戳变得更加困难。例如，考虑一个移动应用程序，它向服务器报告用于使用度量的事件。在设备离线的情况下，应用程序可能被使用，并且在设备上缓冲事件，并在下次可用的互联网连接（可能是几小时甚至几天后）将它们发送到服务器。对于这个流的任何消费者来说，事件将显得非常迟滞。

In this context, the timestamp on the events should really be the time at which the user interaction occurred, according to the mobile device’s local clock. However, the clock on a user-controlled device often cannot be trusted, as it may be accidentally or deliberately set to the wrong time (see “Clock Synchronization and Accuracy” ). The time at which the event was received by the server (according to the server’s clock) is more likely to be accurate, since the server is under your control, but less meaningful in terms of describing the user interaction.

在这种情况下，事件的时间戳应该真正是用户交互发生时的时间，根据移动设备的本地时钟。然而，用户控制的设备上的时钟往往不能被信任，因为它可能会被意外或故意设置为错误的时间（请参见“时钟同步和精度”）。根据服务器时钟接收事件的时间更有可能是准确的，因为服务器是在你的控制之下，但从描述用户交互的角度来看，意义较小。

To adjust for incorrect device clocks, one approach is to log three timestamps [ 82 ]:

调整不正确的设备时钟的一种方法是记录三个时间戳 [82]:

• The time at which the event occurred, according to the device clock

  事件发生的时间，根据设备时钟。

• The time at which the event was sent to the server, according to the device clock

  根据设备时钟发送事件的时间。

• The time at which the event was received by the server, according to the server clock

  服务器接收该事件的时间，根据服务器时钟显示。

By subtracting the second timestamp from the third, you can estimate the offset between the device clock and the server clock (assuming the network delay is negligible compared to the required timestamp accuracy). You can then apply that offset to the event timestamp, and thus estimate the true time at which the event actually occurred (assuming the device clock offset did not change between the time the event occurred and the time it was sent to the server).

通过从第三个时间戳中减去第二个时间戳，可以估计设备时钟和服务器时钟之间的偏移量（假设网络延迟与所需时间戳精度相比可以忽略不计）。然后，可以将该偏移量应用于事件时间戳，从而估计事件实际发生的真正时间（假设设备时钟偏移量在该事件发生时和发送到服务器时没有发生变化）。

This problem is not unique to stream processing—batch processing suffers from exactly the same issues of reasoning about time. It is just more noticeable in a streaming context, where we are more aware of the passage of time.

这个问题不仅仅是流处理所独有的——批量处理在关于时间的推断方面也有完全相同的问题。只是在流处理环境中更加显著，因为我们更加注意时间的流逝。

Types of windows

Once you know how the timestamp of an event should be determined, the next step is to decide how windows over time periods should be defined. The window can then be used for aggregations, for example to count events, or to calculate the average of values within the window. Several types of windows are in common use [ 79 , 83 ]:

一旦确定了事件的时间戳应如何确定，下一步就是决定如何定义时间段内的窗口。接下来可以使用窗口进行聚合，例如计算事件次数或计算窗口内数值的平均值。常用的几种窗口类型包括[79，83]：

Tumbling window

A tumbling window has a fixed length, and every event belongs to exactly one window. For example, if you have a 1-minute tumbling window, all the events with timestamps between 10:03:00 and 10:03:59 are grouped into one window, events between 10:04:00 and 10:04:59 into the next window, and so on. You could implement a 1-minute tumbling window by taking each event timestamp and rounding it down to the nearest minute to determine the window that it belongs to.

滚动窗口具有固定长度，每个事件都属于恰好一个窗口。例如，如果您有一个1分钟的滚动窗口，则所有时间戳介于10:03:00和10:03:59之间的事件都分组为一个窗口，介于10:04:00和10:04:59之间的事件则分为下一个窗口，以此类推。您可以通过将每个事件时间戳向下舍入到最接近的一分钟来实现1分钟的滚动窗口。

Hopping window

A hopping window also has a fixed length, but allows windows to overlap in order to provide some smoothing. For example, a 5-minute window with a hop size of 1 minute would contain the events between 10:03:00 and 10:07:59, then the next window would cover events between 10:04:00 and 10:08:59, and so on. You can implement this hopping window by first calculating 1-minute tumbling windows, and then aggregating over several adjacent windows.

跳跃窗口也有固定的长度，但允许窗口重叠以提供一些平滑。例如，带有1分钟跳跃大小的5分钟窗口将包含10:03:00至10:07:59之间的事件，然后下一个窗口将覆盖10:04:00至10:08:59之间的事件，以此类推。您可以通过先计算1分钟滚动窗口，然后聚合多个相邻窗口来实现此跳跃窗口。

Sliding window

A sliding window contains all the events that occur within some interval of each other. For example, a 5-minute sliding window would cover events at 10:03:39 and 10:08:12, because they are less than 5 minutes apart (note that tumbling and hopping 5-minute windows would not have put these two events in the same window, as they use fixed boundaries). A sliding window can be implemented by keeping a buffer of events sorted by time and removing old events when they expire from the window.

滑动窗口包含在某段时间内发生的所有事件。例如，5分钟的滑动窗口将包括在10:03:39和10:08:12发生的事件，因为它们相隔不到5分钟（请注意，滚动和跳跃的5分钟窗口不会将这两个事件放在同一个窗口中，因为它们使用固定边界）。滑动窗口可以通过保留按时间排序的事件缓冲区来实现，并在它们从窗口中过期时删除旧事件。

Session window

Unlike the other window types, a session window has no fixed duration. Instead, it is defined by grouping together all events for the same user that occur closely together in time, and the window ends when the user has been inactive for some time (for example, if there have been no events for 30 minutes). Sessionization is a common requirement for website analytics (see “GROUP BY” ).

与其他窗口类型不同，会话窗口没有固定的持续时间。相反，它是通过将同一用户在时间上紧密联系的所有事件组合在一起来定义的，当用户一段时间内处于非活动状态时窗口就会结束（例如，如果30分钟没有出现任何事件）。会话化是网站分析的常见要求（请参见“GROUP BY”）。

Stream-stream join (window join)

Say you have a search feature on your website, and you want to detect recent trends in searched-for URLs. Every time someone types a search query, you log an event containing the query and the results returned. Every time someone clicks one of the search results, you log another event recording the click. In order to calculate the click-through rate for each URL in the search results, you need to bring together the events for the search action and the click action, which are connected by having the same session ID. Similar analyses are needed in advertising systems [ 85 ].

假设您的网站有一个搜索功能，并且您想要检测最近搜索的URL趋势。每当有人键入搜索查询时，您记录一个包含查询和返回结果的事件。每当有人点击搜索结果之一时，您记录另一个事件，记录点击情况。为了计算搜索结果中每个URL的点击率，您需要将搜索操作和点击操作的事件结合起来，它们通过具有相同会话ID来连接。类似的分析也需要在广告系统中进行。

The click may never come if the user abandons their search, and even if it comes, the time between the search and the click may be highly variable: in many cases it might be a few seconds, but it could be as long as days or weeks (if a user runs a search, forgets about that browser tab, and then returns to the tab and clicks a result sometime later). Due to variable network delays, the click event may even arrive before the search event. You can choose a suitable window for the join—for example, you may choose to join a click with a search if they occur at most one hour apart.

如果用户放弃搜索，那么单击可能永远不会出现，即使它出现了，在搜索和单击之间的时间也可能高度变化：在许多情况下可能只有几秒钟，但它也可能长达数天或数周（如果用户运行 搜索，忘记了浏览器选项卡，然后稍后返回选项卡并单击结果）。由于网络延迟的不确定性，单击事件甚至可能在搜索事件之前到达。您可以选择合适的窗口进行连接-例如，如果它们在最多一小时内发生，则可以选择将单击与搜索连接起来。

Note that embedding the details of the search in the click event is not equivalent to joining the events: doing so would only tell you about the cases where the user clicked a search result, not about the searches where the user did not click any of the results. In order to measure search quality, you need accurate click-through rates, for which you need both the search events and the click events.

请注意，在单击事件中嵌入搜索详细信息并不等同于加入事件：这样做只会告诉您用户单击搜索结果的情况，而不是用户没有单击任何结果的搜索情况。为了测量搜索质量，您需要精确的点击率，而这需要搜索事件和单击事件两者都需要。

To implement this type of join, a stream processor needs to maintain state : for example, all the events that occurred in the last hour, indexed by session ID. Whenever a search event or click event occurs, it is added to the appropriate index, and the stream processor also checks the other index to see if another event for the same session ID has already arrived. If there is a matching event, you emit an event saying which search result was clicked. If the search event expires without you seeing a matching click event, you emit an event saying which search results were not clicked.

为实现此类连接，流处理器需要维护状态：例如，所有最近一小时发生的事件，按会话 ID 索引。每当搜索事件或点击事件发生时，它将被添加到相应的索引中，流处理器也将检查其他索引，以查看是否已经到达了与相同会话 ID 的另一个事件。如果存在匹配的事件，则发出一条事件，说明哪个搜索结果已被点击。如果搜索事件在您看不到匹配的点击事件的情况下过期，则发出一条事件，说明哪些搜索结果未被点击。

Stream-table join (stream enrichment)

In “Example: analysis of user activity events” ( Figure 10-2 ) we saw an example of a batch job joining two datasets: a set of user activity events and a database of user profiles. It is natural to think of the user activity events as a stream, and to perform the same join on a continuous basis in a stream processor: the input is a stream of activity events containing a user ID, and the output is a stream of activity events in which the user ID has been augmented with profile information about the user. This process is sometimes known as enriching the activity events with information from the database.

在“示例:用户活动事件的分析”（图10-2）中，我们看到了一个批处理作业示例，在此作业中，两个数据集被合并：一组用户活动事件和一个用户配置文件数据库。自然而然的想法是将用户活动事件视为数据流，并在数据流处理器中持续执行相同的合并：输入是包含用户ID的活动事件流，输出是已经被用户配置文件信息扩充的活动事件流。这个过程有时被称为将活动事件与数据库中的信息进行丰富。

To perform this join, the stream process needs to look at one activity event at a time, look up the event’s user ID in the database, and add the profile information to the activity event. The database lookup could be implemented by querying a remote database; however, as discussed in “Example: analysis of user activity events” , such remote queries are likely to be slow and risk overloading the database [ 75 ].

为了执行这个连接操作，流处理需要逐一查看每个活动事件，从数据库中查找该事件的用户ID并将个人资料信息添加到活动事件中。数据库的查找可以通过查询远程数据库来实现；然而，正如在“示例：用户活动事件分析”中所讨论的，这样的远程查询很可能很慢，有过载数据库的风险[75]。

Another approach is to load a copy of the database into the stream processor so that it can be queried locally without a network round-trip. This technique is very similar to the hash joins we discussed in “Map-Side Joins” : the local copy of the database might be an in-memory hash table if it is small enough, or an index on the local disk.

另一种方法是将数据库副本加载到流处理器中，以便可以在本地查询，而不需要网络往返。这种技术非常类似于我们在“Map-Side Joins”中讨论的哈希连接：如果本地数据库的大小足够小，那么它的本地副本可能是在内存中的哈希表，否则可能是本地磁盘上的索引。

The difference to batch jobs is that a batch job uses a point-in-time snapshot of the database as input, whereas a stream processor is long-running, and the contents of the database are likely to change over time, so the stream processor’s local copy of the database needs to be kept up to date. This issue can be solved by change data capture: the stream processor can subscribe to a changelog of the user profile database as well as the stream of activity events. When a profile is created or modified, the stream processor updates its local copy. Thus, we obtain a join between two streams: the activity events and the profile updates.

批处理任务的区别在于，批处理任务使用数据库的某个时间点的快照作为输入，而流处理器是长时间运行的，数据库的内容随着时间会发生变化，因此流处理器的本地副本需要保持最新。这个问题可以通过变更数据捕获来解决：流处理器可以订阅用户配置文件数据库的变更日志以及活动事件流。当配置文件被创建或修改时，流处理器会更新其本地副本。因此，我们得到了两个流的连接：活动事件和配置文件更新。

A stream-table join is actually very similar to a stream-stream join; the biggest difference is that for the table changelog stream, the join uses a window that reaches back to the “beginning of time” (a conceptually infinite window), with newer versions of records overwriting older ones. For the stream input, the join might not maintain a window at all.

流表联接实际上与流流联接非常相似；最大的区别在于对于表changelog流，联接使用窗口回溯到“时间的开始”（概念上的无限窗口），以新版本的记录覆盖旧的记录。对于流输入，联接可能根本不维护窗口。

Table-table join (materialized view maintenance)

Consider the Twitter timeline example that we discussed in “Describing Load” . We said that when a user wants to view their home timeline, it is too expensive to iterate over all the people the user is following, find their recent tweets, and merge them.

考虑我们在“描述负载”中讨论过的Twitter时间线示例。我们说当用户想要查看他们的主页时间线时，遍历用户关注的所有人，找到他们最近的推文并合并它们是太昂贵了。

Instead, we want a timeline cache: a kind of per-user “inbox” to which tweets are written as they are sent, so that reading the timeline is a single lookup. Materializing and maintaining this cache requires the following event processing:

相反，我们想要一个时间轴缓存：一种每个用户“收件箱”，当推文被发送时写入该收件箱，这样阅读时间轴就是单个查找。实现和维护此缓存需要以下事件处理：

• When user u sends a new tweet, it is added to the timeline of every user who is following u .

  当用户 u 发布新推文时，它将被添加到所有正在关注用户 u 的用户的时间线中。

• When a user deletes a tweet, it is removed from all users’ timelines.

  当用户删除一条推文时，它会从所有用户的时间线上删除。

• When user u ₁ starts following user u ₂ , recent tweets by u ₂ are added to u ₁ ’s timeline.

  当用户 u1 开始关注用户 u2 时，u2 的最近推文将被添加到 u1 的时间线中。

• When user u ₁ unfollows user u ₂ , tweets by u ₂ are removed from u ₁ ’s timeline.

  当用户u1取消关注用户u2时，u2的推文会从u1的时间线中移除。

To implement this cache maintenance in a stream processor, you need streams of events for tweets (sending and deleting) and for follow relationships (following and unfollowing). The stream process needs to maintain a database containing the set of followers for each user so that it knows which timelines need to be updated when a new tweet arrives [ 86 ].

为了在流处理器中实现此缓存维护，您需要具有推文事件（发送和删除）和关注关系（关注和取消关注）的事件流。流处理过程需要维护一个包含每个用户的关注者集合的数据库，以便知道在新推文到达时需要更新哪些时间线。[86]。

Another way of looking at this stream process is that it maintains a materialized view for a query that joins two tables (tweets and follows), something like the following:

另一种看待这个流程的方式是它为一个连接两个表（tweets和follows）的查询维护了一个物化视图，类似于以下内容：

SELECT follows.follower_id AS timeline_id,
  array_agg(tweets.* ORDER BY tweets.timestamp DESC)
FROM tweets
JOIN follows ON follows.followee_id = tweets.sender_id
GROUP BY follows.follower_id

The join of the streams corresponds directly to the join of the tables in that query. The timelines are effectively a cache of the result of this query, updated every time the underlying tables change. ⁱⁱⁱ

流的联接直接对应于查询中表的联接。时间轴实际上是这个查询结果的缓存，在基础表变化时每次更新。

Time-dependence of joins

The three types of joins described here (stream-stream, stream-table, and table-table) have a lot in common: they all require the stream processor to maintain some state (search and click events, user profiles, or follower list) based on one join input, and query that state on messages from the other join input.

这里描述的三种连接类型（流-流、流-表和表-表）有许多共同点：它们都需要流处理器基于一个连接输入维护一些状态（搜索和点击事件、用户资料或关注者列表），并查询来自其他连接输入的信息的状态。

The order of the events that maintain the state is important (it matters whether you first follow and then unfollow, or the other way round). In a partitioned log, the ordering of events within a single partition is preserved, but there is typically no ordering guarantee across different streams or partitions.

维护状态的事件顺序很重要（首先关注再取消或者相反，顺序有影响）。在分区日志中，单个分区内的事件顺序被保留，但通常没有跨不同流或分区的顺序保证。

This raises a question: if events on different streams happen around a similar time, in which order are they processed? In the stream-table join example, if a user updates their profile, which activity events are joined with the old profile (processed before the profile update), and which are joined with the new profile (processed after the profile update)? Put another way: if state changes over time, and you join with some state, what point in time do you use for the join [ 45 ]?

这引发了一个问题：如果不同数据流上的事件在相近的时间内发生，它们的处理顺序是什么？在流表联接的例子中，如果用户更新他们的个人资料，哪些活动事件将与旧资料一起联接（在资料更新之前处理），哪些事件将与新资料一起联接（在资料更新之后处理）？换句话说：如果状态随时间变化，并且您与某些状态联接，那么对于联接使用哪个时间点呢？[45]

Such time dependence can occur in many places. For example, if you sell things, you need to apply the right tax rate to invoices, which depends on the country or state, the type of product, and the date of sale (since tax rates change from time to time). When joining sales to a table of tax rates, you probably want to join with the tax rate at the time of the sale, which may be different from the current tax rate if you are reprocessing historical data.

这种时间上的依赖性可能发生在很多地方。例如，如果你销售产品，你需要将适当的税率应用于发票上，这取决于国家或州、产品类型以及销售日期(因为税率会随着时间变化而变化)。当将销售与税率表结合时，你可能想要使用销售时的税率进行结合，这可能与当前的税率不同，特别是当你重新处理历史数据时。

If the ordering of events across streams is undetermined, the join becomes nondeterministic [ 87 ], which means you cannot rerun the same job on the same input and necessarily get the same result: the events on the input streams may be interleaved in a different way when you run the job again.

如果跨流事件的排序是不确定的，则连接将变得非确定性，这意味着您不能在相同的输入上重新运行相同的作业并且必须得到相同的结果：当您再次运行作业时，输入流中的事件可能以不同的方式交织。

In data warehouses, this issue is known as a slowly changing dimension (SCD), and it is often addressed by using a unique identifier for a particular version of the joined record: for example, every time the tax rate changes, it is given a new identifier, and the invoice includes the identifier for the tax rate at the time of sale [ 88 , 89 ]. This change makes the join deterministic, but has the consequence that log compaction is not possible, since all versions of the records in the table need to be retained.

在数据仓库中，这个问题被称为逐渐变化的维度（SCD），通常通过为特定版本的连接记录使用唯一标识符来解决。例如，每当税率发生变化，就会赋予一个新的标识符，并且发票包括销售时税率的标识符[88, 89]。这种改变使得连接是确定性的，但是有一个后果，即日志压缩是不可能的，因为表中所有版本的记录都需要被保留。

Microbatching and checkpointing

One solution is to break the stream into small blocks, and treat each block like a miniature batch process. This approach is called microbatching , and it is used in Spark Streaming [ 91 ]. The batch size is typically around one second, which is the result of a performance compromise: smaller batches incur greater scheduling and coordination overhead, while larger batches mean a longer delay before results of the stream processor become visible.

一种解决方案是将流分成小块，将每个块作为小批量处理。这种方法称为微批处理，它在Spark Streaming [91]中使用。批量大小通常约为一秒钟，这是性能折衷的结果：较小批次会产生更大的调度和协调开销，而较大批次意味着流处理器结果变得可见之前需要更长的延迟。

Microbatching also implicitly provides a tumbling window equal to the batch size (windowed by processing time, not event timestamps); any jobs that require larger windows need to explicitly carry over state from one microbatch to the next.

微批处理还隐含提供一个等于批量大小的滚动窗口（按处理时间而非事件时间戳进行窗口化）；任何需要较大窗口的作业都需要显式地从一个微批处理传递状态到下一个。

A variant approach, used in Apache Flink, is to periodically generate rolling checkpoints of state and write them to durable storage [ 92 , 93 ]. If a stream operator crashes, it can restart from its most recent checkpoint and discard any output generated between the last checkpoint and the crash. The checkpoints are triggered by barriers in the message stream, similar to the boundaries between microbatches, but without forcing a particular window size.

另一种方法是在Apache Flink中使用周期性滚动检查点来存储状态，并将它们写入持久性存储 [92，93]。如果流操作员崩溃，则可以从其最近的检查点重新启动，并丢弃在最后一个检查点和崩溃之间生成的任何输出。检查点是由消息流中的屏障触发的，类似于微批次之间的边界，但不强制任何窗口大小。

Within the confines of the stream processing framework, the microbatching and checkpointing approaches provide the same exactly-once semantics as batch processing. However, as soon as output leaves the stream processor (for example, by writing to a database, sending messages to an external message broker, or sending emails), the framework is no longer able to discard the output of a failed batch. In this case, restarting a failed task causes the external side effect to happen twice, and microbatching or checkpointing alone is not sufficient to prevent this problem.

在流处理框架的范围内，微批处理和检查点方案提供了与批处理完全相同的仅一次语义。但是，一旦输出离开流处理器（例如，写入数据库、发送消息到外部消息代理或发送电子邮件），框架就无法丢弃由于失败的批处理而产生的输出。在这种情况下，重新启动失败的任务会导致外部副作用发生两次，仅仅使用微批处理或检查点方案是不足以解决这个问题的。

Atomic commit revisited

In order to give the appearance of exactly-once processing in the presence of faults, we need to ensure that all outputs and side effects of processing an event take effect if and only if the processing is successful. Those effects include any messages sent to downstream operators or external messaging systems (including email or push notifications), any database writes, any changes to operator state, and any acknowledgment of input messages (including moving the consumer offset forward in a log-based message broker).

为了在故障存在的情况下呈现出精确一次处理的外观，我们需要确保所有事件的处理所产生的输出和副作用只有在处理成功时才会生效。这些影响包括发送到下游操作器或外部消息系统（包括电子邮件或推送通知）的任何消息、任何数据库写入、任何操作器状态更改以及对输入消息的确认（包括将消费者偏移量向前移动在基于日志的消息代理中）。

Those things either all need to happen atomically, or none of them must happen, but they should not go out of sync with each other. If this approach sounds familiar, it is because we discussed it in “Exactly-once message processing” in the context of distributed transactions and two-phase commit.

这些事情要么必须同时发生，要么就不能发生，但它们不应该与彼此失去同步。如果这种方法听起来很熟悉，那是因为我们曾在分布式事务和两阶段提交的“恰好一次消息处理”中讨论过它。

In Chapter 9 we discussed the problems in the traditional implementations of distributed transactions, such as XA. However, in more restricted environments it is possible to implement such an atomic commit facility efficiently. This approach is used in Google Cloud Dataflow [ 81 , 92 ] and VoltDB [ 94 ], and there are plans to add similar features to Apache Kafka [ 95 , 96 ]. Unlike XA, these implementations do not attempt to provide transactions across heterogeneous technologies, but instead keep them internal by managing both state changes and messaging within the stream processing framework. The overhead of the transaction protocol can be amortized by processing several input messages within a single transaction.

第9章中，我们讨论了传统分布式事务（如XA）实现中存在的问题。然而，在更受限制的环境中，可以有效地实现这样的原子提交功能。Google Cloud Dataflow[81，92]和VoltDB [94]使用了这种方法，并计划在Apache Kafka [95,96] 中添加类似功能。与XA不同，这些实现不尝试通过异构技术提供事务，而是通过在流处理框架内管理状态更改和消息来保持它们的内部性。事务协议的开销可以通过在单个事务中处理多个输入消息而摊销。

Idempotence

Our goal is to discard the partial output of any failed tasks so that they can be safely retried without taking effect twice. Distributed transactions are one way of achieving that goal, but another way is to rely on idempotence [ 97 ].

我们的目标是丢弃任何失败任务的部分输出，以便可以安全地重试而不会产生两倍影响。分布式事务是实现这一目标的一种方式，但另一种方式是依靠幂等性。

An idempotent operation is one that you can perform multiple times, and it has the same effect as if you performed it only once. For example, setting a key in a key-value store to some fixed value is idempotent (writing the value again simply overwrites the value with an identical value), whereas incrementing a counter is not idempotent (performing the increment again means the value is incremented twice).

幂等操作是指你可以多次执行的操作，而它的效果和只执行一次一样。例如，在键值存储中将键设置为某个固定值是幂等的（再次写入相同的值只会将该值覆盖），而增加计数器不是幂等的（再次进行增加意味着值增加了两次）。

Even if an operation is not naturally idempotent, it can often be made idempotent with a bit of extra metadata. For example, when consuming messages from Kafka, every message has a persistent, monotonically increasing offset. When writing a value to an external database, you can include the offset of the message that triggered the last write with the value. Thus, you can tell whether an update has already been applied, and avoid performing the same update again.

即使一个操作本质上不是幂等的，加入一些额外的元数据也可以让它变为幂等的。例如，在从Kafka接收消息时，每个消息都有一个持久化的、单调递增的偏移量。当将一个值写入外部数据库时，可以将触发最后一次写入的消息的偏移量与该值一起包含在内。这样，就可以判断一个更新是否已经被应用，避免再次执行相同的更新。

The state handling in Storm’s Trident is based on a similar idea [ 78 ]. Relying on idempotence implies several assumptions: restarting a failed task must replay the same messages in the same order (a log-based message broker does this), the processing must be deterministic, and no other node may concurrently update the same value [ 98 , 99 ].

Storm的Trident中的状态处理基于类似的思路。依赖幂等性意味着有几个假设：如果重新启动失败的任务，则必须以相同的顺序再次播放相同的消息（基于日志的消息代理程序可以实现此操作），处理必须是确定性的，并且不能同时更新同一值的任何其他节点。

When failing over from one processing node to another, fencing may be required (see “The leader and the lock” ) to prevent interference from a node that is thought to be dead but is actually alive. Despite all those caveats, idempotent operations can be an effective way of achieving exactly-once semantics with only a small overhead.

当从一个处理节点故障转移至另一个处理节点时，可能需要进行隔离（请参见“领导者和锁定”），以防止来自被认为已死亡但实际上仍然活动的节点的干扰。尽管有所有这些警告，但幂等操作可以有效地实现精确一次性语义而只带来小的开销。

Rebuilding state after a failure

Any stream process that requires state—for example, any windowed aggregations (such as counters, averages, and histograms) and any tables and indexes used for joins—must ensure that this state can be recovered after a failure.

任何需要状态的流处理过程——例如任何窗口聚合（例如计数器、平均值和直方图）以及用于连接的任何表和索引——必须确保在故障后可以恢复此状态。

One option is to keep the state in a remote datastore and replicate it, although having to query a remote database for each individual message can be slow, as discussed in “Stream-table join (stream enrichment)” . An alternative is to keep state local to the stream processor, and replicate it periodically. Then, when the stream processor is recovering from a failure, the new task can read the replicated state and resume processing without data loss.

一种选择是将状态保持在远程数据仓库中并复制它，尽管每个消息都需要查询远程数据库可能会很慢，如“流-表连接（流增强）”中所讨论的那样。 另一种选择是将状态保持在流处理器本地并定期进行复制。然后，当流处理器恢复失败时，新任务可以读取复制的状态并在没有数据丢失的情况下恢复处理。

For example, Flink periodically captures snapshots of operator state and writes them to durable storage such as HDFS [ 92 , 93 ]; Samza and Kafka Streams replicate state changes by sending them to a dedicated Kafka topic with log compaction, similar to change data capture [ 84 , 100 ]. VoltDB replicates state by redundantly processing each input message on several nodes (see “Actual Serial Execution” ).

例如，Flink定期捕获操作状态的快照并将其写入持久存储，例如HDFS [92，93]；Samza和Kafka Streams通过将状态更改发送到专用Kafka主题并进行日志压缩来复制状态，类似于更改数据捕获 [84，100]. VoltDB通过在多个节点上冗余处理每个输入消息来复制状态（参见“实际串行执行”）。

In some cases, it may not even be necessary to replicate the state, because it can be rebuilt from the input streams. For example, if the state consists of aggregations over a fairly short window, it may be fast enough to simply replay the input events corresponding to that window. If the state is a local replica of a database, maintained by change data capture, the database can also be rebuilt from the log-compacted change stream (see “Log compaction” ).

在某些情况下，甚至不必复制状态，因为它可以从输入流重建。例如，如果状态是在相当短的窗口期内聚合的，那么只需简单地重播与该窗口相对应的输入事件即可快速完成。如果状态是由变更数据捕获维护的数据库的本地副本，则数据库也可以从经过日志压缩的更改流中重建 (见“日志压缩”)。

However, all of these trade-offs depend on the performance characteristics of the underlying infrastructure: in some systems, network delay may be lower than disk access latency, and network bandwidth may be comparable to disk bandwidth. There is no universally ideal trade-off for all situations, and the merits of local versus remote state may also shift as storage and networking technologies evolve.

然而，所有这些权衡取决于底层基础设施的性能特征：在某些系统中，网络延迟可能低于磁盘访问延迟，并且网络带宽可能与磁盘带宽相当。对于所有情况，没有普遍理想的权衡，当存储和网络技术发展时，本地与远程状态的优点也可能会发生变化。