Producer API

消息发送流程

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

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

异步发送 API

1）导入依赖

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

2）编写代码
需要用到的类：
KafkaProducer：需要创建一个生产者对象，用来发送数据
ProducerConfig：获取所需的一系列配置参数
ProducerRecord：每条数据都要封装成一个 ProducerRecord 对象
1.不带回调函数的 API

public class CustomProducer {
	public static void main(String[] args) throws ExecutionException,InterruptedException {
		Properties props = new Properties();
 		//kafka 集群，broker-list
 		props.put("bootstrap.servers", "hadoop102:9092");
 		props.put("acks", "all");
		//重试次数
		props.put("retries", 1);
		//批次大小
		props.put("batch.size", 16384);
		//等待时间
		props.put("linger.ms", 1);
		//RecordAccumulator 缓冲区大小
		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);
		for (int i = 0; i < 100; i++) {
 			producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));
 		}
 		producer.close();
 	}
}

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

public class CustomProducer {
	public static void main(String[] args) throws ExecutionException, InterruptedException {
		Properties props = new Properties();
 		props.put("bootstrap.servers", "hadoop102: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>("first", Integer.toString(i), Integer.toString(i)), new Callback() {
 				//回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用
 				@Override
 				public void onCompletion(RecordMetadata metadata, Exception exception) {
 					if (exception == null) {
 						System.out.println("success->" + metadata.offset());
 					} else {
						exception.printStackTrace();
					}
 				}
 			});
 		}
 		producer.close();
 	}
}

同步发送 API

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

public class CustomProducer {
	public static void main(String[] args) throws ExecutionException, InterruptedException {
 		Properties props = new Properties();
 		props.put("bootstrap.servers", "hadoop102: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>("first", Integer.toString(i), Integer.toString(i))).get();
 		}
 		producer.close();
	 }
}