Kafka事务与高级特性

cheniy2025/8/15大约 3 分钟

Apache Kafka事务与高级特性

前置知识

在学习本文之前，请确保您已经：

了解Kafka基本概念
掌握生产者和消费者的使用
熟悉分布式事务理论

事务机制

1. 事务概述

Kafka事务可以保证：

原子性写入：多条消息要么全部成功，要么全部失败
跨分区事务：支持跨多个分区的事务
读隔离性：消费者可以配置只读取已提交的消息

2. 事务配置

@Data
@Configuration
public class KafkaTransactionConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        
        // 事务相关配置
        config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tx-");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        
        return new DefaultKafkaProducerFactory<>(config);
    }
    
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> template = 
            new KafkaTemplate<>(producerFactory());
        template.setTransactionIdPrefix("tx-");
        return template;
    }
}

3. 事务示例

事务生产者示例代码

@Slf4j
@Service
public class TransactionalProducer {
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    public void sendMessagesInTransaction(
            List<String> messages, String topic) {
        kafkaTemplate.executeInTransaction(operations -> {
            try {
                for (String message : messages) {
                    operations.send(topic, message)
                            .addCallback(
                                result -> log.info("Sent: {}", message),
                                ex -> log.error("Failed to send: {}", message, ex)
                            );
                }
                return true; // 提交事务
            } catch (Exception e) {
                log.error("Transaction failed", e);
                throw e; // 回滚事务
            }
        });
    }
}

4. 事务消费者

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> config = new HashMap<>();
    
    // 设置隔离级别
    config.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, 
              "read_committed");
    
    return new DefaultKafkaConsumerFactory<>(config);
}

幂等性

1. 幂等性概述

幂等性保证

Kafka的幂等性可以保证：

消息不会重复写入
消息顺序不会乱序
适用于单个生产者会话

2. 幂等配置

// 启用幂等性
config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
// 必要的配置
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

安全机制

1. SSL/TLS配置

SSL配置示例

@Configuration
public class KafkaSSLConfig {
    @Bean
    public Map<String, Object> producerSSLConfig() {
        Map<String, Object> config = new HashMap<>();
        
        // SSL配置
        config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SSL");
        config.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, 
                  "client.truststore.jks");
        config.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, 
                  "truststore-password");
        config.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, 
                  "client.keystore.jks");
        config.put(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, 
                  "keystore-password");
        config.put(SslConfigs.SSL_KEY_PASSWORD_CONFIG, 
                  "key-password");
        
        return config;
    }
}

2. SASL认证

// SASL配置
config.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, 
          "SASL_SSL");
config.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
config.put(SaslConfigs.SASL_JAAS_CONFIG, 
          "org.apache.kafka.common.security.plain.PlainLoginModule " +
          "required username=\"alice\" password=\"alice-secret\";");

高级功能

1. 消息压缩

// 配置压缩
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");
// 可选值: none, gzip, snappy, lz4, zstd

2. 消息头部

@Slf4j
public class HeadersExample {
    public void sendWithHeaders(String topic, String message) {
        ProducerRecord<String, String> record = 
            new ProducerRecord<>(topic, null, message);
        
        // 添加头部信息
        record.headers()
              .add("timestamp", 
                   String.valueOf(System.currentTimeMillis())
                         .getBytes())
              .add("source", "web-app".getBytes());
        
        kafkaTemplate.send(record);
    }
    
    public void consumeWithHeaders(ConsumerRecord<String, String> record) {
        for (Header header : record.headers()) {
            log.info("Header: {} = {}", 
                     header.key(), 
                     new String(header.value()));
        }
    }
}