贺州招聘网站建设,网站中竖导航栏怎么做,天津网站建设企业,国外媒体中文网站Apache Kafka 事务详解
Apache Kafka 是一个分布式流处理平台#xff0c;主要用于实时数据的传输和处理。在现代的数据密集型应用中#xff0c;事务性保证在数据传输和处理中的作用至关重要。本文将详细介绍 Kafka 的事务性支持#xff0c;包括其基本概念、架构、使用方法以…Apache Kafka 事务详解
Apache Kafka 是一个分布式流处理平台主要用于实时数据的传输和处理。在现代的数据密集型应用中事务性保证在数据传输和处理中的作用至关重要。本文将详细介绍 Kafka 的事务性支持包括其基本概念、架构、使用方法以及相关代码示例和运行效果。
1. Kafka 事务简介
Kafka 的事务性支持在 0.11.0 版本中引入目的是提供跨多个 topic 和 partition 的原子消息写入能力。这意味着事务消息要么全部写入成功要么全部失败从而确保数据的一致性和完整性。
Kafka 的事务特性主要用于以下场景
确保多个 topic 和 partition 的消息一致性实现端到端的 Exactly Once 语义EOS防止消息丢失或重复消费
2. Kafka 事务架构
Kafka 事务涉及三个主要组件
生产者Producer负责发送事务性消息。消费者Consumer负责消费事务性消息。Kafka Broker负责管理事务状态确保事务的一致性。
在 Kafka 中每个事务都有一个唯一的 Transactional ID用于标识事务的生命周期。事务的状态通过 Broker 中的事务协调器Transaction Coordinator进行管理。
3. Kafka 事务使用方法
3.1 配置生产者
要使用 Kafka 事务性支持首先需要配置生产者。下面是一个配置事务性生产者的示例
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.Properties;
import java.util.concurrent.ExecutionException;public class TransactionalProducer {public static void main(String[] args) throws ExecutionException, InterruptedException {Properties props new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer);props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, my-transactional-id);KafkaProducerString, String producer new KafkaProducer(props);producer.initTransactions();try {producer.beginTransaction();producer.send(new ProducerRecord(my-topic, key1, value1)).get();producer.send(new ProducerRecord(my-topic, key2, value2)).get();producer.commitTransaction();} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {producer.close();throw e;} catch (KafkaException e) {producer.abortTransaction();}producer.close();}
}3.2 配置消费者
为了正确消费事务性消息需要配置隔离级别isolation.level为“读已提交read_committed”
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;import java.util.Collections;
import java.util.Properties;public class TransactionalConsumer {public static void main(String[] args) {Properties props new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, localhost:9092);props.put(ConsumerConfig.GROUP_ID_CONFIG, my-group);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringDeserializer);props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, read_committed);KafkaConsumerString, String consumer new KafkaConsumer(props);consumer.subscribe(Collections.singletonList(my-topic));while (true) {ConsumerRecordsString, String records consumer.poll(100);for (ConsumerRecordString, String record : records) {System.out.printf(offset %d, key %s, value %s%n, record.offset(), record.key(), record.value());}}}
}4. 事务运行效果
4.1 生产者运行效果
当事务性生产者运行时如果事务成功提交我们可以看到以下输出
Successfully sent message: key1, value1
Successfully sent message: key2, value2如果事务失败并被回滚我们将不会看到任何消息发送成功的日志。
4.2 消费者运行效果
事务性消费者只会读取已提交的事务消息。例如如果我们发送了两条消息但只提交了一条那么消费者只会读取已提交的那条消息。
offset 0, key key1, value value1未提交的消息将不会被读取从而确保数据的一致性。
5. 总结
Kafka 的事务性支持提供了一种确保消息一致性和完整性的方法尤其适用于需要跨多个 topic 和 partition 进行原子写入的场景。通过配置事务性生产者和消费者我们可以实现端到端的 Exactly Once 语义防止消息丢失或重复消费。希望本文能帮助你更好地理解和使用 Kafka 的事务特性。
参考文献
Apache Kafka DocumentationConfluent Kafka Transactions
