Kafka producer端开发代码实例

下面是详细的Kafka producer端开发代码实例攻略：

1. 搭建开发环境

首先，需要搭建Kafka的开发环境。可以参考官方文档：http://kafka.apache.org/quickstart。

2. 引入Kafka的依赖库

在Maven项目中，需要引入以下依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

3. 编写Producer端代码

3.1 创建Producer实例

首先，需要创建一个Kafka producer实例，代码如下：

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092"); // 指定kafka broker地址
props.put("acks", "all"); // 确认模式 all 或 1
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);

在创建Kafka producer实例的时候，需要指定Kafka broker的地址、确认模式和序列化器等参数。

3.2 发送消息

使用以下方法发送消息到Kafka broker：

ProducerRecord<String, String> record = new ProducerRecord<>("topic_name", "message_key", "message_value");
producer.send(record);

其中，ProducerRecord类表示待发送的消息，包括消息的主题、键和值。producer.send(record)方法将消息发送到Kafka broker。

3.3 关闭Producer实例

在使用完Kafka producer实例后，需要关闭它以释放资源，代码如下：

producer.close();

4. 示例代码

下面是两条示例代码，分别演示了如何发送不同类型的消息。

4.1 发送字符串消息

public static void sendStringMessage() {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    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);

    String topic = "test_topic";
    String message = "Hello, Kafka!";

    ProducerRecord<String, String> record = new ProducerRecord<>(topic, "key", message);
    producer.send(record);

    producer.close();
}

4.2 发送JSON消息

public static void sendJsonMessage() {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("acks", "all");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "com.example.MyJsonSerializer"); // 自定义JSON序列化器

    KafkaProducer<String, MyJson> producer = new KafkaProducer<>(props);

    String topic = "test_topic";
    MyJson json = new MyJson("name", 18);

    ProducerRecord<String, MyJson> record = new ProducerRecord<>(topic, "key", json);
    producer.send(record);

    producer.close();
}

这里的MyJson是自定义的JSON对象，MyJsonSerializer是自定义的JSON序列化器。需要注意的是，需要将自定义的序列化器的类名放到value.serializer属性中。