预见猿份

Kafka从入门到应用 ​

第2章 生产者详解 ​

tips 学完这一章你可以

  • 深入学习Kafka数据生产大致流程
  • 如何创建并使用Kafka生产者
  • Kafka生产者常用配置

2.1 消息发送 ​

2.1.1 Kafka Java客户端数据生产流程解析 ​

img_106.png

①、首先要构造一个 ProducerRecord 对象，该对象可以声明主题Topic、分区Partition、键 Key以及值 Value，主题和值是必须要声明的，分区和键可以不用指定。

②、调用send() 方法进行消息发送。

③、因为消息要到网络上进行传输，所以必须进行序列化，序列化器的作用就是把消息的 key 和value对象序列化成字节数组。

④、接下来数据传到分区器，如果之间的 ProducerRecord 对象指定了分区，那么分区器将不再做任何事，直接把指定的分区返回；如果没有，那么分区器会根据 Key 来选择一个分区，选择好分区之后，生产者就知道该往哪个主题和分区发送记录了。

⑤、接着这条记录会被添加到一个记录批次里面，这个批次里所有的消息会被发送到相同的主题和分区。会有一个独立的线程来把这些记录批次发送到相应的 Broker 上。

③、Broker成功接收到消息，表示发送成功，返回消息的元数据（包括主题和分区信息以及记录在分区里的偏移量）。发送失败，可以选择重试或者直接抛出异常。

依赖的包 <kafka.version>2.0.0</kafka.version>

text
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${scala.version}</artifactId>
<version>${kafka.version}</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>

2.1.2 必要参数配置 ​

见代码库：com.yjoffer.kafka.chapter2.KafkaProducerAnalysis

text
public static Properties initConfig() {
Properties props = new Properties();
// 该属性指定 brokers 的地址清单，格式为 host:port。清单里不需要包含所有的 broker
地址，
// 生产者会从给定的 broker 里查找到其它 broker 的信息。——建议至少提供两个 broker
的信息，因为一旦其中一个宕机，生产者仍然能够连接到集群上。
props.put("bootstrap.servers", brokerList);
// 将 key 转换为字节数组的配置，必须设定为一个实现了
org.apache.kafka.common.serialization.Serializer 接口的类，
// 生产者会用这个类把键对象序列化为字节数组。
// ——kafka 默认提供了 StringSerializer和 IntegerSerializer、
ByteArraySerializer。当然也可以自定义序列化器。
props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
// 和 key.serializer 一样，用于 value 的序列化
props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer");
text
// 用来设定KafkaProducer对应的客户端ID，默认为空，如果不设置KafkaProducer会自动
生成一个非空字符串。
// 内容形式如："producer-1"
props.put("client.id", "producer.client.id.demo");
return props;
}
text
Properties props = initConfig();
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
//        KafkaProducer<String, String> producer = new KafkaProducer<>(props,
//                new StringSerializer(), new StringSerializer());
//生成 ProducerRecord 对象，并制定 Topic，key 以及 value
ProducerRecord<String, String> record = new ProducerRecord<>(topic,
"hello, Kafka!");
try {
// 发送消息
producer.send(record);

2.13 发送类型 ​

发送即忘记

producer.send(record)

同步发送

text
//通过send()发送完消息后返回一个Future对象,然后调用Future对象的get()方法等待kafka响应
//如果kafka正常响应，返回一个RecordMetadata对象，该对象存储消息的偏移量
// 如果kafka发生错误，无法正常响应，就会抛出异常，我们便可以进行异常处理
producer.send(record).get();

异步发送

text
producer.send(record, new Callback() {
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println(metadata.partition() + ":" + metadata.offset());
}
}
});

2.1.4 序列化器 ​

消息要到网络上进行传输，必须进行序列化，而序列化器的作用就是如此。

Kafka 提供了默认的字符串序列化器（org.apache.kafka.common.serialization.StringSerializer），还有整型（IntegerSerializer）和字节数组（BytesSerializer）序列化器，这些序列化器都实现了接口（org.apache.kafka.common.serialization.Serializer）基本上能够满足大部分场景的需求。

2.1.5 自定义序列化器 ​

见代码库：com.yjoffer.kafka.chapter2.CompanySerializer

text
/**
* 自定义序列化器
text
*/
public class CompanySerializer implements Serializer<Company> {
@Override
public void configure(Map configs, boolean isKey) {
}
text
@Override
public byte[] serialize(String topic, Company data) {
if (data == null) {
return null;
}
byte[] name, address;
try {
if (data.getName() != null) {
name = data.getName().getBytes("UTF-8");
} else {
name = new byte[0];
}
if (data.getAddress() != null) {
address = data.getAddress().getBytes("UTF-8");
} else {
address = new byte[0];
}
ByteBuffer buffer = ByteBuffer.
allocate(4 + 4 + name.length + address.length);
buffer.putInt(name.length);
buffer.put(name);
buffer.putInt(address.length);
buffer.put(address);
return buffer.array();
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
return new byte[0];
}
text
@Override
public void close() {
}
}

使用自定义的序列化器

见代码库：com.yjoffer.kafka.chapter2.ProducerDefineSerializer

text
public class ProducerDefineSerializer {
public static final String brokerList = "localhost:9092";
public static final String topic = "heima";
text
public static void main(String[] args)
throws ExecutionException, InterruptedException {
Properties properties = new Properties();
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
CompanySerializer.class.getName());
//        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
text
//                ProtostuffSerializer.class.getName());
properties.put("bootstrap.servers", brokerList);
text
KafkaProducer<String, Company> producer =
new KafkaProducer<>(properties);
Company company = Company.builder().name("kafka")
.address("北京").build();
//        Company company = Company.builder().name("hiddenkafka")
//                .address("China").telphone("13000000000").build();
ProducerRecord<String, Company> record =
new ProducerRecord<>(topic, company);
producer.send(record).get();
}
}

2.1.6 分区器 ​

本身kafka有自己的分区策略的，如果未指定，就会使用默认的分区策略：

Kafka根据传递消息的key来进行分区的分配，即hash(key) % numPartitions。如果Key相同的话，那么就会分配到统一分区。

源代码org.apache.kafka.clients.producer.internals.DefaultPartitioner分析

text
public int partition(String topic, Object key, byte[] keyBytes, Object value,
byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
if (keyBytes == null) {
int nextValue = this.nextValue(topic);
List<PartitionInfo> availablePartitions =
cluster.availablePartitionsForTopic(topic);
if (availablePartitions.size() > 0) {
int part = Utils.toPositive(nextValue) %
availablePartitions.size();
return
((PartitionInfo)availablePartitions.get(part)).partition();
} else {
return Utils.toPositive(nextValue) % numPartitions;
}
} else {
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
}

自定义分区器见代码库 com.yjoffer.kafka.chapter2.DefinePartitioner

text
/**
* 自定义分区器
*/
public class DefinePartitioner implements Partitioner {
private final AtomicInteger counter = new AtomicInteger(0);
text
@Override
public int partition(String topic, Object key, byte[] keyBytes,
text
Object value, byte[] valueBytes, Cluster cluster) {
List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
int numPartitions = partitions.size();
if (null == keyBytes) {
return counter.getAndIncrement() % numPartitions;
} else
return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> configs) {
}
}

实现自定义分区器需要通过配置参数ProducerConfig.PARTITIONER_CLASS_CONFIG来实现

text
// 自定义分区器的使用
props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getNam
e());

2.1.7 拦截器 ​

Producer拦截器(interceptor)是个相当新的功能，它和consumer端interceptor是在Kafka 0.10版本被引入的，主要用于实现clients端的定制化控制逻辑。

生产者拦截器可以用在消息发送前做一些准备工作。

使用场景

1、按照某个规则过滤掉不符合要求的消息

2、修改消息的内容

3、统计类需求

见代码库：自定义拦截器com.yjoffer.kafka.chapter2.ProducerInterceptorPrefix

text
/**
* 自定义拦截器
*/
public class ProducerInterceptorPrefix implements
ProducerInterceptor<String, String> {
private volatile long sendSuccess = 0;
private volatile long sendFailure = 0;
text
@Override
public ProducerRecord<String, String> onSend(
ProducerRecord<String, String> record) {
String modifiedValue = "prefix1-" + record.value();
return new ProducerRecord<>(record.topic(),
record.partition(), record.timestamp(),
record.key(), modifiedValue, record.headers());
//        if (record.value().length() < 5) {
//            throw new RuntimeException();
text
//        }
//        return record;
}
text
@Override
public void onAcknowledgement(
RecordMetadata recordMetadata,
Exception e) {
if (e == null) {
sendSuccess++;
} else {
sendFailure++;
}
}
text
@Override
public void close() {
double successRatio = (double) sendSuccess / (sendFailure +
sendSuccess);
System.out.println("[INFO] 发送成功率="
+ String.format("%f", successRatio * 100) + "%");
}
text
@Override
public void configure(Map<String, ?> map) {
}
}

实现自定义拦截器之后需要在配置参数中指定这个拦截器，此参数的默认值为空，如下：

text
// 自定义拦截器使用
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerDefineSerializer.cla
ss.getName());

功能演示：

发送端

img_134.png

接收端

img_136.png

2.2 发送原理剖析 ​

img_141.png

消息发送的过程中，涉及到两个线程协同工作，主线程首先将业务数据封装成ProducerRecord对象，之后调用send()方法将消息放入RecordAccumulator(消息收集器，也可以理解为主线程与Sender线程直接的缓冲区)中暂存，Sender线程负责将消息信息构成请求，并最终执行网络I/O的线程，它从RecordAccumulator中取出消息并批量发送出去，需要注意的是，KafkaProducer是线程安全的，多个线程间可以共享使用同一个KafkaProducer对象

2.3 其他生产者参数 ​

之前提及的默认三个客户端参数，大部分参数都有合理的默认值，一般情况下不需要修改它们，

参考官网：http://kafka.apache.org/documentation/#producerconfigs

2.3.1 acks ​

这个参数用来指定分区中必须有多少个副本收到这条消息，之后生产者才会认为这条消息时写入成功的。acks是生产者客户端中非常重要的一个参数，它涉及到消息的可靠性和吞吐量之间的权衡。

ack=0， 生产者在成功写入消息之前不会等待任何来自服务器的相应。如果出现问题生产者是感知不到的，消息就丢失了。不过因为生产者不需要等待服务器响应，所以它可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。ack=1，默认值为1，只要集群的首领节点收到消息，生产这就会收到一个来自服务器的成功响应。如果消息无法达到首领节点（比如首领节点崩溃，新的首领还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。但是，这样还有可能会导致数据丢失，如果收到写成功通知，此时首领节点还没来的及同步数据到follower节点，首领节点崩溃，就会导致数据丢失。ack=-1， 只有当所有参与复制的节点都收到消息时，生产这会收到一个来自服务器的成功响应，这种模式是最安全的，它可以保证不止一个服务器收到消息。

注意：acks参数配置的是一个字符串类型，而不是整数类型，如果配置为整数类型会抛出以下异常

img_148.png

2.3.2 retries ​

生产者从服务器收到的错误有可能是临时性的错误（比如分区找不到首领）。在这种情况下，如果达到

text
了retires 设置的次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待
100ms，可以通过retry.backoff.ms 参数来修改这个时间间隔。

2.3.3 batch.size ​

当有多个消息要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算，而不是消息个数。当批次被填满，批次里的所有消息会被发送出去。不过生产者并不一定都会等到批次被填满才发送，半满的批次，甚至只包含一个消息的批次也可能

text
被发送。所以就算把batch.size 设置的很大，也不会造成延迟，只会占用更多的内存而已，如果设置

的太小，生产者会因为频繁发送消息而增加一些额外的开销。

2.3.4 max.request.size ​

该参数用于控制生产者发送的请求大小，它可以指定能发送的单个消息的最大值，也可以指单个请求里

text
所有消息的总大小。broker 对可接收的消息最大值也有自己的限制（message.max.size ），所以两
边的配置最好匹配，避免生产者发送的消息被broker 拒绝。

总结 ​

本章主要讲了生产者客户端的用法以及整体流程架构，主要内容包括配置参数的详解、消息的发送方式、序列化器、分区器、拦截器等，在实际使用中，Kafka已经提供了良好的Java客户端支持，提高了开发效率。