事务的场景分析
- 最简单的需求是producer发的多条消息组成一个事务这些消息需要对consumer同时可见或者同时不可见 。
- producer可能会给多个topic，多个partition发消息，这些消息也需要能放在一个事务里面，这就形成了一个典型的分布式事务。
- kafka的应用场景经常是应用先消费一个topic，然后做处理再发到另一个topic，这个consume-transform-produce过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费位点也不能提交。
- producer或者producer所在的应用可能会挂掉，新的producer启动以后需要知道怎么处理之前未完成的事务 。
- 流式处理的拓扑可能会比较深，如果下游只有等上游消息事务提交以后才能读到，可能会导致rt非常长吞吐量也随之下降很多，所以需要实现read committed和read uncommitted两种事务隔离级别。
关键概念以及推导
- 因为producer发送消息可能是分布式事务，所以引入了常用的两阶段提交协议，所以有事务协调者(Transaction Coordinator)。Transaction Coordinator和之前为了解决脑裂和惊群问题引入的Group Coordinator在选举和failover上面类似。
- 事务管理中事务日志是必不可少的，kafka使用一个内部topic来保存事务日志，这个设计和之前使用内部topic保存位点的设计保持一致。事务日志是Transaction Coordinator管理的状态的持久化，因为不需要回溯事务的历史状态，所以事务日志只用保存最近的事务状态。
- 3.因为事务存在commit和abort两种操作，而客户端又有read committed和read uncommitted两种隔离级别，所以消息队列必须能标识事务状态，这个被称作Control Message。
- 4.producer挂掉重启或者漂移到其它机器需要能关联的之前的未完成事务所以需要有一个唯一标识符来进行关联，这个就是TransactionalId，一个producer挂了，另一个有相同TransactionalId的producer能够接着处理这个事务未完成的状态。注意不要把TransactionalId和数据库事务中常见的transaction id搞混了，kafka目前没有引入全局序，所以也没有transaction id，这个TransactionalId是用户提前配置的。
- 5. TransactionalId能关联producer，也需要避免两个使用相同TransactionalId的producer同时存在，所以引入了producer epoch来保证对应一个TransactionalId只有一个活跃的producer epoch
事务语义
多分区原子写入
????????事务能够保证Kafka topic下每个分区的原子写入。事务中所有的消息都将被成功写入或者丢弃。例如，处理过程中发生了异常并导致事务终止，这种情况下，事务中的消息都不会被Consumer读取。
????????Kafka是如何实现原子的“读取-处理-写入”过程的。“读取-处理-写入”过程：如果某个应用程序在某个topic tp0的偏移量X处读取到了消息A，并且在对消息A进行了一些处理（如B = F（A））之后将消息B写入topic tp1，则只有当消息A和B被认为被成功地消费并一起发布，或者完全不发布时，整个读取过程写入操作是原子的。
????????现在，只有当消息A的偏移量X被标记为消耗时，消息A才被认为是从topic tp0消耗的，消费到的数据偏移量（record offset）将被标记为提交偏移量（Committing offset）。在Kafka中，我们通过写入一个名为offsets topic的内部Kafka topic来记录offset commit。消息仅在其offset被提交给offsets topic时才被认为成功消费。
????????由于offset commit只是对Kafkatopic的另一次写入，并且由于消息仅在提交偏移量时被视为成功消费，所以跨多个主题和分区的原子写入也启用原子“读取-处理-写入”循环：提交偏移量X到offset topic和消息B到tp1的写入将是单个事务的一部分，所以整个步骤都是原子的。
粉碎“僵尸实例”
????????我们通过为每个事务Producer分配一个称为transactional.id的唯一标识符来解决僵尸实例的问题。在进程重新启动时能够识别相同的Producer实例。
????????API要求事务性Producer的第一个操作应该是在Kafka集群中显示注册transactional.id。 当注册的时候，Kafka broker用给定的transactional.id检查打开的事务并且完成处理。 Kafka也增加了一个与transactional.id相关的epoch。Epoch存储每个transactional.id内部元数据。
????????一旦这个epoch被触发，任何具有相同的transactional.id和更旧的epoch的Producer被视为僵尸，并被围起来, Kafka会拒绝来自这些Procedure的后续事务性写入。
读事务消息
????????Kafka Consumer只有在事务实际提交时才会将事务消息传递给应用程序。也就是说，Consumer不会提交作为整个事务一部分的消息，也不会提交属于中止事务的消息。
????????值得注意的是，上述保证不足以保证整个消息读取的原子性，当使用Kafka consumer来消费来自topic的消息时，应用程序将不知道这些消息是否被写为事务的一部分，因此他们不知道事务何时开始或结束；此外，给定的Consumer不能保证订阅属于事务一部分的所有Partition，并且无法发现这一点，最终难以保证作为事务中的所有消息被单个Consumer处理。
????????简而言之：Kafka保证Consumer最终只能提供非事务性消息或提交事务性消息。它将保留来自未完成事务的消息，并过滤掉已中止事务的消息。
事务处理API
prouder提供事务的api
KafkaProducer.initTransactions() // 事务初始化
KafkaProducer.beginTransaction() // 开始事务
KafkaProducer.send() // 发送数据
KafkaProducer.sendOffsetsToTransaction() //发送offset到消费组协调者
KafkaProducer.commitTransaction() // 提交事务
KafkaProducer.abortTransaction() // 终止事务
api分类
????????在一个原子操作中，根据包含的操作类型，可以分为三种情况，前两种情况是事务引入的场景，最后一种情况没有使用价值。
- 只有Producer生产消息；
- 消费消息和生产消息并存，这个是事务场景中最常用的情况，就是我们常说的“consume-transform-produce ”模式
- 只有consumer消费消息，这种操作其实没有什么意义，跟使用手动提交效果一样，而且也不是事务属性引入的目的，所以一般不会使用这种情况
事务配置
- 创建消费者代码：
- 将配置中的自动提交属性（auto.commit）进行关闭
- 而且在代码里面也不能使用手动提交commitSync( )或者commitAsync( )
- 设置isolation.level
- 创建生成者代码：
- 配置transactional.id属性
- 配置enable.idempotence属性
应用程序实例
只有生产者的事务demo
package com.example.demo.transaction;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Properties;
import java.util.concurrent.Future;
public class TransactionProducer {
private static Properties getProps(){
Properties props = new Properties();
props.put("bootstrap.servers", "47.52.199.53:9092");
props.put("retries", 2); // 重试次数
props.put("batch.size", 100); // 批量发送大小
props.put("buffer.memory", 33554432); // 缓存大小，根据本机内存大小配置
props.put("linger.ms", 1000); // 发送频率，满足任务一个条件发送
props.put("client.id", "producer-syn-2"); // 发送端id,便于统计
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("transactional.id","producer-1"); // 每台机器唯一
props.put("enable.idempotence",true); // 设置幂等性
return props;
}
public static void main(String[] args) {
KafkaProducer<String, String> producer = new KafkaProducer<>(getProps());
// 初始化事务
producer.initTransactions();try {
Thread.sleep(2000);
// 开启事务
producer.beginTransaction();
// 发送消息到producer-syn
producer.send(new ProducerRecord<String, String>("producer-syn","test3"));
// 发送消息到producer-asyn
Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<String, String>("producer-asyn","test4"));
// 提交事务
producer.commitTransaction();
}catch (Exception e){
e.printStackTrace();
// 终止事务
producer.abortTransaction();
}
}
}
消费和生产并存的demo
package com.example.demo.transaction;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Future;
public class consumeTransformProduce {
private static Properties getProducerProps(){
Properties props = new Properties();
props.put("bootstrap.servers", "47.52.199.51:9092");
props.put("retries", 3); // 重试次数
props.put("batch.size", 100); // 批量发送大小
props.put("buffer.memory", 33554432); // 缓存大小，根据本机内存大小配置
props.put("linger.ms", 1000); // 发送频率，满足任务一个条件发送
props.put("client.id", "producer-syn-2"); // 发送端id,便于统计
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("transactional.id","producer-2"); // 每台机器唯一
props.put("enable.idempotence",true); // 设置幂等性
return props;
}
private static Properties getConsumerProps(){
Properties props = new Properties();
props.put("bootstrap.servers", "47.52.199.51:9092");
props.put("group.id", "test_3");
props.put("session.timeout.ms", 30000); // 如果其超时，将会可能触发rebalance并认为已经死去，重新选举Leader
props.put("enable.auto.commit", "false"); // 开启自动提交
props.put("auto.commit.interval.ms", "1000"); // 自动提交时间
props.put("auto.offset.reset","earliest"); // 从最早的offset开始拉取，latest:从最近的offset开始消费
props.put("client.id", "producer-syn-1"); // 发送端id,便于统计
props.put("max.poll.records","100"); // 每次批量拉取条数
props.put("max.poll.interval.ms","1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("isolation.level","read_committed"); // 设置隔离级别
return props;
}
public static void main(String[] args) {
// 创建生产者
KafkaProducer<String, String> producer = new KafkaProducer<>(getProducerProps());
// 创建消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(getConsumerProps());
// 初始化事务
producer.initTransactions();
// 订阅主题
consumer.subscribe(Arrays.asList("consumer-tran"));
for(;;){
// 开启事务
producer.beginTransaction();
// 接受消息
ConsumerRecords<String, String> records = consumer.poll(500);
// 处理逻辑
try {
Map<TopicPartition, OffsetAndMetadata> commits = new HashMap<>();
for(ConsumerRecord record : records){
// 处理消息
System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());
// 记录提交的偏移量
commits.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset()));
// 产生新消息
Future<RecordMetadata> metadataFuture = producer.send(new ProducerRecord<>("consumer-send",record.value()+"send"));
}
// 提交偏移量
producer.sendOffsetsToTransaction(commits,"group0323");
// 事务提交
producer.commitTransaction();
}catch (Exception e){
e.printStackTrace();
producer.abortTransaction();
}
}
}
}
事务工作原理
事务协调器和事务日志
- 事务Coordinator：每个KafkaBroker内部运行的一个模块，主要负责分配pid，记录事务状态等操作。
- 事务日志：事务日志是一个内部的Kafka Topic。事务日志只保存事务的最新状态而不是事务中的实际消息。消息只存储在实际的Topic的分区中。事务可以处于诸如“Ongoing”，“prepare commit”和“Completed”之类的各种状态中。正是这种状态和关联的元数据存储在事务日志中。
事务交互情况
- producer和事务coordinator的交互
- 执行事务时，Producer向事务协调员发出如下请求：initTransactions
- initTransactions API向coordinator注册一个transactional.id（由生产者指定）。 coordinator首先使用该transactional.id关闭所有待处理的事务，并且会避免遇到僵尸实例，由具有相同的transactional.id的Producer的另一个实例启动的任何事务将被关闭和隔离。每个Producer会话只发生一次。
- 当Producer在事务中第一次将数据发送到分区时，首先向coordinator注册分区。
- 当应用程序调用commitTransaction或abortTransaction时，会向coordinator发送一个请求以开始两阶段提交协议。
- Coordinator和事务日志交互
- 随着事务的进行，Producer发送上面的请求来更新Coordinator上事务的状态。事务Coordinator会在内存中保存每个事务的状态，并且把这个状态写到事务日志中（这是以三种方式复制的，因此是持久保存的）
- 事务Coordinator是读写事务日志的唯一组件。如果一个给定的Borker故障了，一个新的Coordinator会被选为新的事务日志的Leader，这个事务日志分割了这个失效的代理，它从传入的分区中读取消息并在内存中重建状态。
- Producer将数据写入目标Topic所在分区
- 在Coordinator的事务中注册新的分区后，Producer将数据正常地发送到真实数据所在分区。这与producer.send流程完全相同，但有一些额外的验证，以确保Producer不被隔离。
- Topic分区和Coordinator的交互
- 在Producer发起提交（或中止）之后，协调器开始两阶段提交协议。
- 在第一阶段，Coordinator将其内部状态更新为“prepare_commit”并在事务日志中更新此状态。一旦完成了这个事务，无论发生什么事，都能保证事务完成。
- Coordinator然后开始阶段2，在那里它将事务提交标记写入作为事务一部分的Topic分区。
- 这些事务标记不会暴露给应用程序，但是在read_committed模式下被Consumer使用来过滤掉被中止事务的消息，并且不返回属于开放事务的消息（即那些在日志中但没有事务标记与他们相关联）。
- 一旦标记被写入，事务协调器将事务标记为“完成”，并且Producer可以开始下一个事务。
事务的流程
- 查找事务协调者（Transaction Coordiantor）
- 首先从本地缓存中查找事务协调者，如果没有的话，那么生产者向任意一个borker发送FindCoordinatorRequest请求来获取Transaction Coordinator的地址
- 初始化事务（initTransaction）
- 在知道事务协调者后，生产者需要往协调者发送初始化pid请求(initPidRequest)，获取pid
- 这个请求分为两种：
- 不带transactionID：这种情况下直接生成一个新的produce ID即可，返回给客户端
- 带transactionID：这种情况下，kafka根据transactionalId获取对应的PID，这个对应关系<TransactionId,pid>是保存在事务日志中
- 恢复（Commit或Abort）之前的Producer未完成的事务
- 对PID对应的epoch进行递增，这样可以保证同一个app的不同实例对应的PID是一样，而epoch是不同的
- 开始事务beginTransaction
- 执行Producer的beginTransacion()，它的作用是Producer在本地记录下这个transaction的状态为开始状态
- 这个操作并没有通知Transaction Coordinator，因为Transaction Coordinator只有在Producer发送第一条消息后才认为事务已经开启
- consume-transform-produce 事务循环：
- 增加分区到事务请求：当生产者有新分区要写入数据，则会发送AddPartitionToTxnRequest到事务协调者。协调者会处理请求，主要做的事情是更新事务元数据信息，并把信息写入到事务日志中（事务Topic）。如果该<Topic, Partition>为该事务中第一个<Topic, Partition>，Transaction Coordinator还会启动对该事务的计时（每个事务都有自己的超时时间
- 生产请求：生产者通过调用send接口发送数据到分区，这些请求新增pid，epoch和sequence number字段。Transaction Coordinator会将该<Transaction, Topic, Partition>存于Transaction Log内，并将其状态置为BEGIN
- 增加消费offset到事务：生产者通过新增的snedOffsets ToTransaction接口，会发送某个分区的Offset信息到事务协调者。协调者会把分区信息增加到事务中
- 事务提交offset请求：当生产者调用事务提交offset接口后，会发送一个TxnOffsetCommitRequest请求到消费组协调者，消费组协调者会把offset存储在__consumer-offsets Topic中。协调者会根据请求的PID和epoch验证生产者是否允许发起这个请求。 消费offset只有当事务提交后才对外可见。
- 事务提交或终结 commitTransaction/abortTransaction
- 当生产者完成事务后，客户端需要显式调用结束事务或者回滚事务。前者会使得消息对消费者可见，后者会对生产数据标记为Abort状态，使得消息对消费者不可见。
- 第一阶段：无论是提交或者回滚，都是发送一个EndTnxRequest请求到事务协调者，写入PREPARE_COMMIT或者PREPARE_ABORT信息到事务记录日志中
- 写数据：事务协调者向事务中每个TopicPartition的Leader发送WriteTxnMarkerRequest请求。每个Broker收到请求后会写入COMMIT(PID)或者ABORT(PID)控制信息到数据日志中。
- 这个信息用于告知消费者当前消息是哪个事务，消息是否应该接受或者丢弃。而对于未提交消息，消费者会缓存该事务的消息直到提交或者回滚。
- 注意：如果事务也涉及到__consumer_offsets，即该事务中有消费数据的操作且将该消费的Offset存于__consumer_offsets中，Transaction Coordinator也需要向该内部Topic的各Partition的Leader发送WriteTxnMarkerRequest从而写入COMMIT(PID)或COMMIT(PID)控制信息
- 第二阶段：当提交和回滚信息写入数据日志后，事务协调者会往事务日志中写入最终的提交或者终止信息以表示事务已经完成，此时大部分于事务有关系的消息都可以被删除（通过标记后面在日志压缩时会被移除），我们只需要保留事务ID以及其时间戳即可
事务相关配置
- Broker configs
- transactional.id.timeout.ms：在ms中，事务协调器在生产者TransactionalId提前过期之前等待的最长时间，并且没有从该生产者TransactionalId接收到任何事务状态更新。默认是604800000(7天)。这允许每周一次的生产者作业维护它们的id
- max.transaction.timeout.ms：事务允许的最大超时。如果客户端请求的事务时间超过此时间，broke将在InitPidRequest中返回InvalidTransactionTimeout错误。这可以防止客户机超时过大，从而导致用户无法从事务中包含的主题读取内容。默认值为900000(15分钟)。这是消息事务需要发送的时间的保守上限
- transaction.state.log.replication.factor：事务状态topic的副本数量。默认值:3
- transaction.state.log.num.partitions：事务状态主题的分区数。默认值:50
- transaction.state.log.min.isr：事务状态主题的每个分区ISR最小数量。默认值:2
- transaction.state.log.segment.bytes：事务状态主题的segment大小。默认值:104857600字节
- Producer configs
- enable.idempotence：开启幂等
- transaction.timeout.ms：事务超时时间。事务协调器在主动中止正在进行的事务之前等待生产者更新事务状态的最长时间。这个配置值将与InitPidRequest一起发送到事务协调器。如果该值大于max.transaction.timeout。在broke中设置ms时，请求将失败，并出现InvalidTransactionTimeout错误。默认是60000。这使得交易不会阻塞下游消费超过一分钟，这在实时应用程序中通常是允许的。
- transactional.id：用于事务性交付的Transactional id。这支持跨多个生产者会话的可靠性语义，因为它允许客户端确保使用相同TransactionalId的事务在启动任何新事务之前已经完成。如果没有提供TransactionalId，则生产者仅限于幂等交付。
- Consumer configs
- isolation.level：
- read_uncommitted:以偏移顺序使用已提交和未提交的消息。
- read_committed:仅以偏移量顺序使用非事务性消息或已提交事务性消息。为了维护偏移排序，这个设置意味着我们必须在使用者中缓冲消息，直到看到给定事务中的所有消息。
事务性能以及如何优化
- Producer打开事务之后的性能：事务会造成中等的写入放大。
- 额外的写入在于：
- 对于每个事务，我们都有额外的RPC向Coordinator注册分区
- 在完成事务时，必须将一个事务标记写入参与事务的每个分区。同样，事务Coordinator在单个RPC中批量绑定到同一个Borker的所有标记，所以我们在那里保存RPC开销。但是在事务中对每个分区进行额外的写操作是无法避免的。
- 最后，我们将状态更改写入事务日志。这包括写入添加到事务的每批分区，“prepare_commit”状态和“complete_commit”状态。
- 我们可以看到，开销与作为事务一部分写入的消息数量无关。所以拥有更高吞吐量的关键是每个事务包含更多的消息。
- 实际上，对于Producer以最大吞吐量生产1KB记录，每100ms提交消息导致吞吐量仅降低3％。较小的消息或较短的事务提交间隔会导致更严重的降级。
- 增加事务时间的主要折衷是增加了端到端延迟。回想一下，Consum阅读事务消息不会传递属于公开传输的消息。因此，提交之间的时间间隔越长，消耗的应用程序就越需要等待，从而增加了端到端的延迟。
- Consumer打开之后的性能：
- Consumer在开启事务的场景比Producer简单得多，它需要做的是：
- 过滤掉属于中止事务的消息。
- 不返回属于公开事务一部分的事务消息。
- 因此，当以read_committed模式读取事务消息时，事务Consumer的吞吐量没有降低。这样做的主要原因是我们在读取事务消息时保持零拷贝读取。
- 此外，Consumer不需要任何缓冲等待事务完成。相反，Broker不允许提前抵消包括公开事务。
- 因此，Consumer是非常轻巧和高效的。
|