pulsar事务

之前pulsar消息机制，和架构概览作为一个后端搬砖的需要了解的也差不多了。再补充个pulsar事务

因为exactly-once语义应用场景很多

pulsar事务可以能使流应用程序能够在一个原子操作中 ，消费、处理消息，生成消息。

需要事务的原因

随着流处理星期，对具有更强的处理保障的流处理应用的需求也随之增长。很多业务场景，例如金融，使用流处理引擎为用户处理借方和贷方的业务。这类场景无一例外的要求消息每条消息都只被处理一次。换句话说，如果流处理应用程序消费消息A并生成结果作为消息B（B=f(A)）则 exactly-once 处理意味着当且仅当B成功生成，反之全部失败

Pulsar事务API增强了流处理和消息传递语义和处理保证。他使得流处理应用程序 能够在一个原子操作中使用、处理和生成消息。这意味着事务中 的一批消息可以从多个主题分区接收、生成并由多个主题分区确认。事务中涉及的所有操作作为一个单元成功或失败 。

幂等生产者的限制

使用pulsasr幂等生产者可以避免数据丢失或者重复，但他不能为跨多个分区 的写入提供保证（多分区topic保证不了呗）

在pulsar中，最高级别的消息传递保证是在单分区使用具有exactly once语义的幂等生产者。即每条消息都只持久化u一次 ，而不会丢失数据和重复。但这个方案会有一些限制:

• 由于单调递增的seq ID （理解为消息序列号） 这种方案只能用于单分区，单生产者。多分区多生产者是没有原子性的

  如果在生产消息或接受消息出些异常（某些组件 crash,例如client，broker, bk等）消息会重新发送/处理，可能会造成消息丢失或重复

  • 对于生产者，你也不知道这次丢了哪些数据，会尝试重新发送一定量数据，这会造成数据被持久化了多次 。如果不重发，那么有些数据是持久化了一次，但其他的就会丢了

  • 对于消费者来说，他不知道broker是否受到了消息（ack消息） 因此消费者可能不会重试发送ack，这会导致收到重复消息。

    所以消费者 需要依赖更多机制保证ACK一次

什么是pulsar事务

事务增强了pullsar的消息传递语义和pulsar function的处理保证。pulsar事务API支持多个主题的原子写入和确认。

事务允许：

• 生产者将一批消息发送到多个topic，其中该批次中的所有消息要么最终对消费者都可见，要么都不可见
• 端到端 exactly-once语义(消费，处理，生产)

事务语义

pulsar事务有如下语义

• 所有操作在事务内都会做当作一个单元提交
  • 要么所有消息提交，要么一个不提交
  • 每个而消息被写入或消费一次，不会丢失数据或重复，即使发生故障
  • 如果事务被中止，所有事务内的写入和ack都会回滚
• 事务中的一组消息可以从多个分区接收，生成，并由多个分区确认
  • 消费者只能读取已提交的消息。换句话说，broker不会传递属于事务中的消息或者属于中止事务的消息
  • 跨多分区消息写入是原子的
  • 跨多个subscripitions的ack是原子的。在一个事务ID中ACK消息时，一个消息被成功确认仅一次

事务和流处理

流处理在pulsar中就是一个消费—处理—生产操作。

pulsar事务支持端到端exatly-once流处理。意味着消息不会在source处丢失和在sink处重复

用例

pulsar2.8.0之前，流处理APP不好构建exactyly-once。例如pulsar + flink。pulsar flink connector在puslar 2.8.0之前只提供at-least-once sink connector 和 exactly-once source connector。这意味着最多支持at-least-once语义。在2.8.0后，pulsar flink sink connector 可以提供exactly-onnce语义，基于实现TwoPhaseCommitSinkFunction并且用pulsar事务API和sink消息关联起来。

事务如何工作的

关键概念

TC

transaction coordinator事务协调器是一个broker里的模块。

• 它维护了事务的整个生命周期 ，和保护事务不发生错误
• 处理事务超时，确保事务超时后终止

Transaction log

所有事务元数据都保留在事务日志中。事务日志由pulsar topic支持 。如果TC崩溃，他可以从 事务日志中 恢复事务的元数据。事务日志存储的是事务的状态而不是 事务中实际的消息（实际消息存储在 实际的主题分区中）

Transaction buffer

事务内向主题分区生成的消息会被存储在事务缓冲区（TB）中。在提交事务前，事务缓冲区的消息对消费者都是 不可见的。事务中止时，事务缓冲区 的消息都会被丢弃

事务缓冲区将所有正在进行和中止的事务存储在 内存中。所有消息都发送到实际的topic中。事务提交后，事务缓冲区中的消息对消费者而言可见。

Transaction ID

TxnID 是表示pulsar中唯一事务。事务ID是128bit大小。最高16位用于保留TC的ID，其余用于每个TC中单调递增的数字。通过 TxnId 可以轻松定位到异常的事务

Pending acknowledge state

待确认状态管理在事务完成之前维护事务内消息的ACK。如果消息处于待确认状态，则该消息不能被其他事务确认，直到该消息从挂起确认状态中 删除

pending ack state 是保存在pending acknowledge log中。新的broker可以从日志中 恢复对应状态，确保ACK不会丢失。

数据流

开启事务

Step	Description
1.1	pulsar client 找到对应的TC
1.2	transaction coordinator会为事务分配事务ID。在事务日志中，记录事务及其事务ID和状态（OPEN）。这确保无论TC如何崩溃，事务状态都会持久保存
1.3	事务日志发送保存事务ID的 结果给TC
1.4	记录事务状态条目后，tc把TxnId发回给pulsar client

事务内推送消息

在这个阶段。pulsar client 进入事务循环，重复消费-处理-生产操作。这是个的阶段。可能由多个生成和确认请求组成。

事务中，推送消息步骤如下

Step	Description
2.1.1	在向新主题分区发送消息前。需要发送请求给TC区将对应分区添加到事务中
2.1.2	TC将事务分区的修改记录到事务日志中确保持久性。这确保TC直到事务正在处理的所有分区。TCK而已在分区结束阶段提交或中止每个分区上的修改
2.1.3	事务日志将分区修改结果发送回给TC
2.1.4	TC返回添加新分区到事务的结果给client
2.2.1	pulsasr客户端开始 向分区生成消息。这部分流程与正常消息相同，芝士事务生成的消息包含事务ID
2.2.2	broker往分区记录消息

事务中ack消息

在这个阶段,client向TC发送请求，将订阅名确认为事务的一部分

Step	Description
3.1.1	client发送请求给TC 区添加已经确认的订阅名
3.1.2	TC记录订阅的添加，这确保他直到事务处理的所有订阅。可以在结束阶段提交或者中止每个订阅的更改
3.1.3	事务日志将记录新分区的结果（用于确认消息），发给TC
3.1.4	TC响应clinet的请求，发送被确认的新分区添加的结果
3.2	pulsar客户端确认订阅上的消息 ，这部分请求和确认消息流程相同，芝士确认请求携带事务ID
3.3	broker 接收到确认请求并校验是否属于事务

事务结束阶段

在结束阶段,pulsar client决定提交或中止事务。当确认 消息中检测到冲突时，可以中止事务

End transaction request

当客户端完成了一个事务，需要发送一个结束事务请求

Step	Description
4.1.1	客户端发送结束事务请求（里面有个字段代表是事务提交还是中止）.
4.1.2	transaction coordinator记录COMMITTING 或 ABORTING 到事务日志中
4.1.3	事务日志发送提交或中止记录的结果

事务完成阶段

该阶段，TC会向事务中的所有分区提交或中 修改

Step	Description
4.2.1	TC在订阅上提交事务，和在分区上提交事务
4.2.2	broker（producer）将实际的提交写入实际分区里。broker(ack)将acked的commited mark写入订阅 的pending ack 分区
4.2.3	数据日志将写入产生的提交发给broker。同时 pending ack data log 将写入acked已提交的结果发给broker。cursor移动到下一个位置

标记事务提交或中止

TC会将最终事务状态写到事务日志完成事务

Step	Description
4.3.1	在成功提交或中止此次事务中涉及的所有分区所生产的消息和ACKH偶。TC会将最终COMMITTED或ABORTED事务状态写入事务日志。表明事务完成。事务日志中与该事务关联的所有消息都可以安全删除
4.3.2	事务日志将已提交的事务结果返回给TC
4.3.3	发送事务提交结果给client

总的来说事务是指消费，处理，生产操作。其中就涉及到上面的ACK阶段和推送消息阶段

https://cloud.tencent.com/developer/article/1949452