本篇博主带来的是Kafka的Producer API操作。

1. 消息发送流程

Kafka的Producer发送消息采用的是异步发送的方式。在消息发送的过程中，涉及到了两个线程——main线程和Sender线程，以及一个线程共享变量——RecordAccumulator。main线程将消息发送给RecordAccumulator，Sender线程不断从RecordAccumulator中拉取消息发送到Kafka broker。

下图为KafkaProducer发送消息流程

相关参数：
batch.size：只有数据积累到batch.size之后，sender才会发送数据。
linger.ms：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据。

2. 无回调参数的API

1. 导入依赖

 <build> <plugins> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <configuration> <source>8</source> <target>8</target> </configuration> </plugin> </plugins> </build> <dependencies> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.2</version> </dependency> </dependencies>

2. 编写代码

需要用到的类：
KafkaProducer：需要创建一个生产者对象，用来发送数据
ProducerConfig：获取所需的一系列配置参数
ProducerRecord：每条数据都要封装成一个ProducerRecord对象

3. 完整代码

package com.buwenbuhuo.kafka.producer;

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;
import java.util.concurrent.ExecutionException;


/**
 * @author 卜温不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer { public static void main(String[] args) throws ExecutionException, InterruptedException { Properties props = new Properties(); props.put("bootstrap.servers", "hadoop002:9092");//kafka集群，broker-list props.put("acks", "all"); props.put("retries", 1);//重试次数 props.put("batch.size", 16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小 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); for (int i = 0; i < 100; i++) { producer.send(new ProducerRecord<String, String>("second", i + "", "my name is bu wen bu huo -" + i)); } producer.close(); }
}

4. 群起zookeeper和kafka

[bigdata@hadoop002 zookeeper-3.4.10]$ bin/start-allzk.sh 
[bigdata@hadoop002 kafka]$ bin/start-kafkaall.sh

5. 启动一个consumer控制台

[bigdata@hadoop002 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop002:9092 --topic first

6. 跑代码进行测试

为什么是无序的，这是因为我们只能保证分区内有序。其他无序。

3. 带回调函数的API

回调函数会在producer收到ack时调用，为异步调用，该方法有两个参数，分别是RecordMetadata和Exception，如果Exception为null，说明消息发送成功，如果Exception不为null，说明消息发送失败。
注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

1. 代码

package com.buwenbuhuo.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


/**
 * @author 卜温不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer { public static void main(String[] args) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop002:9092"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.put(ProducerConfig.ACKS_CONFIG, "all"); props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); props.put(ProducerConfig.LINGER_MS_CONFIG, 1); //1.创建1个生产者对象 KafkaProducer<String, String> producer = new KafkaProducer<>(props); //2.调用send方法 for (int i = 0; i < 1000; i++) { producer.send(new ProducerRecord<String, String>("second", i + "", "message-" + i), (metadata,exception)->{ if(exception==null){ System.out.println("success"+metadata.topic()+"-"+metadata.partition()+"-"+metadata.offset()); }else{ exception.printStackTrace(); } }); } //3.关闭生产者 producer.close(); }
}

2. 结果

3. 大致流程图

4. 同步发送API / 只是比异步多了一个.get()

同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回ack。
由于send方法返回的是一个Future对象，根据Futrue对象的特点，我们也可以实现同步发送的效果，只需在调用Future对象的get方发即可。

1. 代码：

package com.buwenbuhuo.kafka.producer;

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author 卜温不火
 * @create 2020-05-06 20:21
 * com.buwenbuhuo.kafka.producer - the name of the target package where the new class or interface will be created.
 * kafka0506 - the name of the current project.
 */
public class CustomProducer { public static void main(String[] args) throws ExecutionException, InterruptedException { Properties props = new Properties(); props.put("bootstrap.servers", "hadoop002:9092");//kafka集群，broker-list props.put("acks", "all"); props.put("retries", 1);//重试次数 props.put("batch.size", 16384);//批次大小 props.put("linger.ms", 1);//等待时间 props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小 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); for (int i = 0; i < 100; i++) { producer.send(new ProducerRecord<String, String>("second", i + "", "my name is bu wen bu huo -" + i)).get(); } producer.close(); }
}

2. 结果

本次的分享就到这里了,

$\color{#FF0000}{看完就赞，养成习惯！！！}$ ^ _ ^ ❤️ ❤️ ❤️
码字不易，大家的支持就是我坚持下去的动力。点赞后不要忘了关注我哦！

文章来源: buwenbuhuo.blog.csdn.net，作者：不温卜火，版权归原作者所有，如需转载，请联系作者。

原文链接：buwenbuhuo.blog.csdn.net/article/details/105957020

（完）

Kafka快速入门系列(9) | Kafka的Producer API操作

目录

1. 消息发送流程

2. 无回调参数的API

3. 带回调函数的API

4. 同步发送API / 只是比异步多了一个.get()