分布式 - 消息队列Kafka：Kafka生产者发送消息的方式

在Kafka中，生产者客户端负责将消息发送到Kafka集群。以下是使用Java Kafka客户端发送消息的示例代码：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
 
public class KafkaProducerExample {
    public static void main(String[] args) {
        // 配置Kafka生产者
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092"); // Kafka集群地址
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 
        Producer<String, String> producer = new KafkaProducer<>(props);
        try {
            // 发送消息
            for (int i = 0; i < 100; i++) {
                String message = "Message " + i;
                producer.send(new ProducerRecord<>("test-topic", message), (metadata, exception) -> {
                    if (exception == null) {
                        System.out.println("Message sent successfully. Topic: " + metadata.topic() + ", Partition: " + metadata.partition() + ", Offset: " + metadata.offset());
                    } else {
                        exception.printStackTrace();
                    }
                });
            }
        } finally {
            // 关闭生产者
            producer.close();
        }
    }
}

在这个示例中，我们配置了一个Kafka生产者，然后在循环中发送了100条消息到名为"test-topic"的主题。每条消息都是通过ProducerRecord对象指定的，包括目标主题和要发送的内容。我们还提供了一个回调函数，该函数在消息成功发送或失败时被调用，以便进行相应的处理。最后，我们关闭了生产者以释放资源。