Kafka?在?java?中的基本使用步驟

更新時間：2025年09月05日 10:55:38 作者：張小虎在學(xué)習(xí)

本文給大家介紹Kafka在java中的基本使用,本文分步驟結(jié)合實例代碼給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友參考下吧

1. 使用 kafka 原生客戶端

現(xiàn)在基本都直接使用 springboot 版本，但了解原生客戶端，能更好的理解 springboot 版的 kafka 客戶端原理。

步驟1：pom 引入核心依賴：

引入依賴時，盡量選擇和 kafka 版本對應(yīng)的依賴版本。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.13</artifactId>
    <version>4.0.0</version>
</dependency>

步驟2：提供者客戶端代碼：

提供者客戶端要做三件事：

設(shè)置提供者客戶端屬性（可選屬性都被定義在 ProducerConfig 類中）
設(shè)置要發(fā)送的消息
發(fā)送（有三種發(fā)送方式，下面代碼中都有）

public class MyProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 第一步：設(shè)置提供者屬性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.28:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            // 第二步：設(shè)置要發(fā)送的消息
            ProducerRecord<String, String> record = new ProducerRecord<>("testTopic", "testKey", "testValue");
            // 第三部：發(fā)送消息
            // send(producer, record);
            // sendSync(producer, record);
            sendASync(producer, record);
        }
    }
    /**
     * 發(fā)送方式1:單向推送，不關(guān)心服務(wù)器的應(yīng)答
     */
    private static void send(Producer<String, String> producer, ProducerRecord<String, String> record) {
        producer.send(record);
    }
    /**
     * 發(fā)送方式2:同步推送，得到服務(wù)器的應(yīng)答前會阻塞當(dāng)前線程
     */
    private static void sendSync(Producer<String, String> producer, ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {
        RecordMetadata metadata = producer.send(record).get();
        System.out.println(metadata.topic());
        System.out.println(metadata.partition());
        System.out.println(metadata.offset());
    }
    /**
     * 發(fā)送方式3:異步推送，不需等待服務(wù)器應(yīng)答，當(dāng)服務(wù)器有應(yīng)答后會觸發(fā)函數(shù)回調(diào)
     */
    private static void sendASync(Producer<String, String> producer, ProducerRecord<String, String> record) {
        producer.send(record, (metadata, exception) -> {
            if (exception != null) {
                throw new RuntimeException("向 kafka 推送失敗", exception);
            }
            System.out.println(metadata.topic());
            System.out.println(metadata.partition());
            System.out.println(metadata.offset());
        });
    }
}

步驟3：消費(fèi)者客戶端代碼：

消費(fèi)者客戶端要做三件事：

設(shè)置消費(fèi)者客戶端屬性（可選屬性都被定義在 ConsumerConfig 類中）
設(shè)置消費(fèi)者訂閱的主題
拉取消息
提交 offset（有兩種提交方式，下面代碼中都有）

public class MyConsumer {
    public static void main(String[] args) {
        // 第一步：設(shè)置消費(fèi)者屬性
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.2.28:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "testGroup");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // 第二步：設(shè)置要訂閱的主題
            consumer.subscribe(Collections.singletonList("testTopic"));
            while (true) {
                // 第三步：拉取消息，100 代表最大等待時間，如果時間到了還沒有拉取到消息就不阻塞了繼續(xù)往后執(zhí)行
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(record.value());
                }
                // 第四步：提交 offset
                // consumer.commitSync(); // 同步提交，表示必須等到 offset 提交完畢，再去消費(fèi)下?批數(shù)據(jù)
                consumer.commitSync(); // 異步提交，表示發(fā)送完提交 offset 請求后，就開始消費(fèi)下?批數(shù)據(jù)了。不?等到Broker的確認(rèn)。
            }
        }
    }
}

2. Kafka 集成 springboot

springboot 版本是最常用的，比原生客戶端使用方便。但是道理是一樣的，底層也是原生客戶端。

pom 引入核心依賴：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.1.0</version>
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
</dependencies>

yaml 配置文件：

這一步無非就是把原生客戶端中的屬性配置，寫在 yaml 中

spring:
  kafka:
    bootstrap-servers: 192.168.2.28:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: testGroup
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

提供者客戶端代碼：

只需要兩步：

注入 KafkaTemplate
發(fā)送

@RestController
public class ProducerController {
    /**
     * kafka
     */
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Autowired
    public void setKafkaTemplate(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }
    @GetMapping("/test")
    public void send() {
        // 發(fā)送 kafka 消息
        kafkaTemplate.send("testTopic", "testKey", "testValue");
    }
}

消費(fèi)者客戶端代碼：

只需要監(jiān)聽主題就可以

@RestController
public class ConsumerController {
    // 監(jiān)聽 kafka 消息
    @KafkaListener(topics = {"testTopic"})
    public void test(ConsumerRecord<?, ?> record) {
        System.out.println(record.value());
    }
}

到此這篇關(guān)于Kafka 在 java 中的基本使用的文章就介紹到這了,更多相關(guān)Kafka java使用內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: