<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.6.0</version> <!-- 版本按需選，建議用較新穩(wěn)定版 -->
    </dependency>
</dependencies>

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducerDemo {
    public static void main(String[] args) {
        // 1. 配置 Kafka 連接、序列化等參數(shù)
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 鍵的序列化器
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 值的序列化器
        // 2. 創(chuàng)建 Producer 實例
        Producer<String, String> producer = new KafkaProducer<>(props);
        // 3. 構造消息（指定主題、鍵、值）
        String topic = "test_topic"; // 要發(fā)送到的主題，需提前在 Kafka 創(chuàng)建或允許自動創(chuàng)建
        String key = "key1";
        String value = "Hello, Kafka from IDEA!";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        // 4. 發(fā)送消息（異步發(fā)送 + 回調處理結果）
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("消息發(fā)送失?。? + exception.getMessage());
                } else {
                    System.out.printf("消息發(fā)送成功！主題：%s，分區(qū)：%d，偏移量：%d%n", 
                        metadata.topic(), metadata.partition(), metadata.offset());
                }
            }
        });
        // 5. 關閉 Producer（實際生產環(huán)境可能在程序結束時或合適時機關閉）
        producer.close();
    }
}

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class KafkaProducerExample {
    private final static String TOPIC = "mytopic";
    private final static String BOOTSTRAP_SERVERS = "localhost:9092";
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        try {
            for (int i = 0; i < 10; i++) {
                String message = "Message " + i;
                producer.send(new ProducerRecord<>(TOPIC, message));
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerDemo {
    public static void main(String[] args) {
        // 1. 配置 Kafka 連接、反序列化、消費者組等參數(shù)
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka 集群地址
        props.put("group.id", "test_group"); // 消費者組 ID，同一組內消費者協(xié)調消費
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 鍵的反序列化器
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); // 值的反序列化器
        props.put("auto.offset.reset", "earliest"); // 沒有已提交偏移量時，從最早消息開始消費
        // 2. 創(chuàng)建 Consumer 實例
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 3. 訂閱主題
        String topic = "test_topic";
        consumer.subscribe(Collections.singletonList(topic));
        // 4. 循環(huán)拉取消息（長輪詢）
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("收到消息：主題=%s，分區(qū)=%d，偏移量=%d，鍵=%s，值=%s%n", 
                        record.topic(), record.partition(), record.offset(), 
                        record.key(), record.value());
                }
                // 手動提交偏移量（也可配置自動提交，生產環(huán)境建議手動更可靠）
                consumer.commitSync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 5. 關閉 Consumer
            consumer.close();
        }
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
    private final static String TOPIC = "mytopic";
    private final static String BOOTSTRAP_SERVERS = "localhost:9092";
    private final static String GROUP_ID = "mygroup";
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        props.put("group.id", GROUP_ID);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                // 處理接收到的消息
                records.forEach(record -> {
                    System.out.println("Received message: " + record.value());
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}