Kafka：生产者发送消息数据去重

你的名字 2024-03-25 21:08 85阅读 0赞

###### **幂等性** ######

开启参数 **enable.idempotence** 默认为 true，false 关闭。

###### **生产者事务** ######

**开启事务，必须开启幂等性。**

Kafka 的事务一共有如下 5 个 API

// 1 初始化事务
    void initTransactions();
    // 2 开启事务
    void beginTransaction() throws ProducerFencedException;
    // 3 在事务内提交已经消费的偏移量（主要用于消费者）
    void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets, String consumerGroupId) throws ProducerFencedException;
    // 4 提交事务
    void commitTransaction() throws ProducerFencedException;
    // 5 放弃事务（类似于回滚事务的操作）
    void abortTransaction() throws ProducerFencedException;

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.common.serialization.StringSerializer;
    
    import java.util.Properties;
    
    public class CustomProducerTranactions {
    
        public static void main(String[] args) {
    
            // 0 配置
            Properties properties = new Properties();
    
            // 连接集群 bootstrap.servers
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.25.129:9092,192.168.25.128:9092");
    
            // 指定对应的key和value的序列化类型 key.serializer
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    
            // 设置事务 id（必须），事务 id 任意起名 "tranactional_id_01"随意确保全局唯一即可
            properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "tranactional_id_01");
    
            // 1 创建kafka生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
    
            kafkaProducer.initTransactions();
    
            kafkaProducer.beginTransaction();
    
            try {
                // 2 发送数据
                for (int i = 0; i < 5; i++) {
                    kafkaProducer.send(new ProducerRecord<>("test", "hello" + i));
                }
    
                int i = 1 / 0;
    
                kafkaProducer.commitTransaction();
            } catch (Exception e) {
                kafkaProducer.abortTransaction();
            } finally {
                // 3 关闭资源
                kafkaProducer.close();
            }
        }
    }