一. 生产者消息发送流程
在消息发送的过程中，涉及到了两个线程：main线程和Sender线程。Producer发送的消息会分别经过Interceptors(拦截器)，Serializer(序列化器)，Partitioner(分区器)最终到达RecordAccumulator，RecordAccumulator是一个双端队列，主要起缓冲区的作用。Sender线程不断从RecordAccumulator中拉取消息发送到 Kafka集群。
二. 异步发送
1. 普通异步发送
普通异步发送指生产者在完成消息发送后不会等待Kafka集群的响应，而是继续去发送下一条消息。
代码实现：
package kafka;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;public class KafkaSenderDemo {private final static String BOOTSTRAP_SERVERS = "192.168.205.154:9092,192.168.205.155:9092,192.168.205.156:9092";public static void main(String[] args) throws Exception {Properties properties = new Properties();// 设置bootstrap server地址properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);// 设置消息的key和value的序列化器properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 10; i++) {producer.send(new ProducerRecord<String, String>("first", "message: " + i));}producer.close();}
}
maven依赖
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.6.0</version>
</dependency>
启动kafka-console-consumer消费者
[root@hadoop1 kafka-3.6.0]# ./bin/kafka-console-consumer.sh --bootstrap-server 192.168.205.154:9092 --topic first
运行结果：
2. 带回调的异步发送
带回调的异步发送是指异步发送后，生产者收到Kafka集群返回的Ack时会执行回调函数。
代码实现：
package kafka;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;public class KafkaSenderDemo {private final static String BOOTSTRAP_SERVERS = "192.168.205.154:9092,192.168.205.155:9092,192.168.205.156:9092";public static void main(String[] args) throws Exception {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 10; i++) {producer.send(new ProducerRecord<String, String>("first", String.valueOf(i), "message: " + i), new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (e == null) {System.out.println("主题:" + recordMetadata.topic() + ", 分区: " + recordMetadata.partition());} else {e.printStackTrace();}}});}producer.close();}
}
运行结果：
主题:first, 分区: 1
主题:first, 分区: 1
主题:first, 分区: 0
主题:first, 分区: 0
主题:first, 分区: 0
主题:first, 分区: 0
主题:first, 分区: 2
主题:first, 分区: 2
主题:first, 分区: 2
主题:first, 分区: 2
三. 同步发送
在同步发送模式下，生产者发送完消息后会阻塞等待Kafka集群的响应，生产者收到Kafka集群的Ack才会进行下一步操作，同步发送的方式大大提高了消息的可靠性，但是也会因此损失性能。同步发送只需要执行完send方法后再调用一下 get()方法即可。
代码实现：
package kafka;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties;
import java.util.concurrent.ExecutionException;public class KafkaSenderDemo {private final static String BOOTSTRAP_SERVERS = "192.168.205.154:9092,192.168.205.155:9092,192.168.205.156:9092";public static void main(String[] args) throws Exception {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 10; i++) {producer.send(new ProducerRecord<String, String>("first", "sync message: " + i)).get();}producer.close();}
}
运行结果：
四. 自定义分区器
Kafka中的Topic是可以分区的，使用分区的好处是显而易见的，它可以合理的使用存储资源，提高并行度，一个Topic的多个分区分散在不同的主机上，可以充分利用集群资源。
生产者发送的每一条消息最终只会进入某一个分区，决定消息和分区映射关系的就是Partitioner。Kafka默认分区器是DefaultPartitioner。
以下是DefaultPartitioner的部分源代码：
public class DefaultPartitioner implements Partitioner {
...public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {return partition(topic, key, keyBytes, value, valueBytes, cluster, cluster.partitionsForTopic(topic).size());}public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster,int numPartitions) {if (keyBytes == null) {return stickyPartitionCache.partition(topic, cluster);}return BuiltInPartitioner.partitionForKey(keyBytes, numPartitions);}
...
}
根据源码可以得出DefaultPartitioner的映射规则如下：
- 指明partition的情况下，直接将指明的值作为partition的值。
- 没有指明partition但有key的情况下，将key的hash值与topic的分区数取余得到partition的值。
- 既没有指明key又没有partition值的情况下，kafka采用Sticky Partition(粘性分区器)，会随机选择一个分区，并一致尽可能使用该分区，待该分区的batch已满或者已完成，kafka再随机选择一个分区进行使用(和上一次的分区不同)
如果DefaultPartitioner不能满足实际业务的分区要求，那么可以自定义分区器，要自定义分区器只需要实现Partitioner类即可。下面以一个需求来说明如何实现自定义分区器。
需求：
将包含hello字符串的消息发送到分区0
将包含world字符串的消息发送到分区1
代码实现：
KafkaOwnPartitionerDemo.java
package kafka;import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;public class KafkaOwnPartitionerDemo {private final static String BOOTSTRAP_SERVERS = "192.168.205.154:9092,192.168.205.155:9092,192.168.205.156:9092";public static void main(String[] args) {Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);for (int i = 0; i < 10; i++) {String[] msgPrefix = {"hello", "world"};String msg = msgPrefix[i % 2] + i;producer.send(new ProducerRecord<String, String>("first", String.valueOf(i), msg), new Callback() {@Overridepublic void onCompletion(RecordMetadata recordMetadata, Exception e) {if (e == null) {System.out.println("消息:" + msg + ", 主题: " + recordMetadata.topic() + ", 分区: " + recordMetadata.partition());} else {e.printStackTrace();}}});}producer.close();}
}MyPartitioner.java
package kafka;import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import java.util.Map;public class MyPartitioner implements Partitioner {@Overridepublic int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {String msg = new String(valueBytes);if (msg.contains("hello")) {return 0;} else {return 1;}}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> configs) {}
}
运行结果：
消息:hello0, 主题: first, 分区: 0
消息:hello2, 主题: first, 分区: 0
消息:hello4, 主题: first, 分区: 0
消息:hello6, 主题: first, 分区: 0
消息:hello8, 主题: first, 分区: 0
消息:world1, 主题: first, 分区: 1
消息:world3, 主题: first, 分区: 1
消息:world5, 主题: first, 分区: 1
消息:world7, 主题: first, 分区: 1
消息:world9, 主题: first, 分区: 1
五. 生产者重要参数列表
| 参数名称 | 参数描述 |
|---|---|
| bootstrap.servers | 生 产 者 连 接 集 群 所 需 的 broker 地 址 清 单 |
| key.serializer 和 value.serializer | 指定发送消息的key和value的序列化类型 |
| buffer.memory | RecordAccumulator 缓冲区总大小，默认 32m。 |
| batch.size | 缓冲区一批数据最大值， 默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。 |
| linger.ms | 如果数据迟迟未达到 batch.size， sender 等待 linger.time之后就会发送数据。单位 ms， 默认值是 0ms，表示没有延迟。 |
| acks | - 0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据， Leader 收到数据后应答。-1（all）：生产者发送过来的数据， Leader+和 ISR队列里面的所有节点收齐数据后应答 |
