快捷導航

java發(fā)送kafka事務(wù)消息的實現(xiàn)方法

更新時間：2022年07月15日 09:52:58 作者：逆風飛翔的小叔

本文主要介紹了java發(fā)送kafka事務(wù)消息的實現(xiàn)方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

前言

事務(wù)對java開發(fā)的同學來說并不陌生，我們使用事務(wù)的目的在于避免產(chǎn)生重復(fù)數(shù)據(jù)或者說利用數(shù)據(jù)存儲中間件的事務(wù)特性確保數(shù)據(jù)的精準性，比如大家熟悉的mysql，我們在程序開始時，只需要在程序中添加上事務(wù)注解即可

kafka客戶端事務(wù)，直接使用客戶端提供的相關(guān)的API即可，和jdbc事務(wù)的使用很類似，主要包含下面5個API

// 1 初始化事務(wù)
void initTransactions();


// 2 開啟事務(wù)
void beginTransaction() throws ProducerFencedException;


// 3 在事務(wù)內(nèi)提交已經(jīng)消費的偏移量（主要用于消費者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 String consumerGroupId) throws ProducerFencedException;


// 4 提交事務(wù)
void commitTransaction() throws ProducerFencedException;


// 5 放棄事務(wù)（類似于回滾事務(wù)的操作）
void abortTransaction() throws ProducerFencedException;

下面結(jié)合實際的代碼以及效果演示進行說明

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.util.Properties;
 
public class ProducerTransaction {
 
    public static void main(String[] args) throws Exception {
 
        // 1. 創(chuàng)建 kafka 生產(chǎn)者的配置對象
        Properties properties = new Properties();
        // 2. 給 kafka 配置對象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "IP:9092");
 
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
 
        // 設(shè)置事務(wù) id（必須），事務(wù) id 任意起名
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
 
        // 3. 創(chuàng)建 kafka 生產(chǎn)者對象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
 
        // 初始化事務(wù)
        kafkaProducer.initTransactions();
        // 開啟事務(wù)
        kafkaProducer.beginTransaction();
        System.out.println("開始發(fā)送消息");
        try {
            // 4. 調(diào)用 send 方法,發(fā)送消息
            for (int i = 0; i < 5; i++) {
                // 發(fā)送消息
                kafkaProducer.send(new ProducerRecord<>("zcy222", "hello kafka " + i));
            }
            //int i = 1 / 0;
            // 提交事務(wù)
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            System.out.println(e);
            // 終止事務(wù)
            kafkaProducer.abortTransaction();
        } finally {
            // 5. 關(guān)閉資源
            kafkaProducer.close();
        }
    }
 
}

運行上面的代碼，正常是可以發(fā)送到指定的topic下