Kafka生产者详解

一.消息发送

1.java客户端数据生产流程解析

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

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

③ 因为消息要到网络上进行传输，所以必须进行序列化，序列化器的作用就是把消息的 key 和value对象序列化成字节数组后，生产者就知道该往哪个主题和分区发送记录了。

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

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

依赖的maven包

        <kafka.version>2.0.0</kafka.version><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.必要参数配置

    private static final String brokerList = "192.168.37.129:9092";private static final String topic = "test";
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");// 用来设定KafkaProducer对应的客户端ID，默认为空，如果不设置KafkaProducer会自动生成一个非空字符串。// 内容形式如："producer-1"props.put("client.id", "producer.client.id.demo");return props;}
public static void main(String[] args) throws InterruptedException {Properties props = initConfig();KafkaProducer<String, String> producer = new KafkaProducer<>(props);ProducerRecord<String, String> record = new ProducerRecord<>(topic,  "hello, Kafka!");try {// 1、发送消息producer.send(record);} catch (Exception e) {e.printStackTrace();}producer.close();}

3.发送类型

发送即忘记

producer.send(record);

同步发送

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

异步发送

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

4.序列化器

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

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

5.自定义序列化器

/*** 自定义序列化器*/
public class CompanySerializer implements Serializer<Company> {@Overridepublic void configure(Map configs, boolean isKey) {}
@Overridepublic 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];}
@Overridepublic void close() {}
}

使用自定义的序列化器

public class ProducerDefineSerializer {public static final String brokerList = "192.168.37.129:9092";public static final String topic = "test";
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("bootstrap.servers", brokerList);
KafkaProducer<String, Company> producer =new KafkaProducer<>(properties);Company company = Company.builder().name("kafka").address("北京").build();ProducerRecord<String, Company> record =new ProducerRecord<>(topic, company);producer.send(record).get();}
}

6.分区器

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

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

源码如下：

我们可以自定义一个分区器

/*** 自定义分区器*/
public class DefinePartitioner implements Partitioner {private final AtomicInteger counter = new AtomicInteger(0);
@Overridepublic 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 (null == keyBytes) {return counter.getAndIncrement() % numPartitions;} else {return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;}}
@Overridepublic void close() {}
@Overridepublic void configure(Map<String, ?> configs) {}
}

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

props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,DefinePartitioner.class.getName());

7.拦截器

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

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

使用场景

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

2）修改消息的内容

3）统计类需求

自定义一个拦截器

/*** 自定义拦截器*/
public class ProducerInterceptorPrefix implementsProducerInterceptor<String, String> {private volatile long sendSuccess = 0;private volatile long sendFailure = 0;
@Overridepublic 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();
//        }
//        return record;}
@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata,Exception e) {if (e == null) {sendSuccess++;} else {sendFailure++;}}
@Overridepublic void close() {double successRatio = (double) sendSuccess / (sendFailure + sendSuccess);System.out.println("[INFO] 发送成功率="+ String.format("%f", successRatio * 100) + "%");}
@Overridepublic void configure(Map<String, ?> map) {}
}

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

props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,ProducerInterceptorPrefix.class.getName());

功能演示：

生产者：

消费者：

二.其它生产者参数

之前提及的默认三个客户端参数，大部分参数都有合理的默认值，一般情况下不需要修改它们，参考官网：http://kafka.apache.org/documentation/#producerconfigs

1.acks

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

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

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

props.put(ProducerConfig.ACKS_CONFIG,"1");

2.retries

生产者从服务器收到的错误有可能是临时性的错误（比如分区找不到首领）。在这种情况下，如果达到了 retires 设置的次数，生产者会放弃重试并返回错误。默认情况下，生产者会在每次重试之间等待100ms，可以通过 retry.backoff.ms 参数来修改这个时间间隔。

3.batch.size

当有多个消息要被发送到同一个分区时，生产者会把它们放在同一个批次里。该参数指定了一个批次可以使用的内存大小，按照字节数计算，而不是消息个数。当批次被填满，批次里的所有消息会被发送出去。不过生产者并不一定都会等到批次被填满才发送，半满的批次，甚至只包含一个消息的批次也可能被发送。所以就算把 batch.size 设置的很大，也不会造成延迟，只会占用更多的内存而已，如果设置的太小，生产者会因为频繁发送消息而增加一些额外的开销。

4.max.request.size

该参数用于控制生产者发送的请求大小，它可以指定能发送的单个消息的最大值，也可以指单个请求里所有消息的总大小。 broker 对可接收的消息最大值也有自己的限制（ message.max.size ），所以两边的配置最好匹配，避免生产者发送的消息被 broker 拒绝。

5.buffer.memory

该参数用来设置生产者内存缓冲区的大小，生产者用它缓冲要发送到服务器的消息。如果应用程序发送消息的速度超过发送到服务器的速度，会导致生产者空间不足。这个时候，send（）方法调用要么被阻塞，要么抛出异常。取决于如何设置 block.on.buffer.full 参数。

6.linger.ms

该参数指定了生产者在发送批次之前等待更多消息加入批次的时间。KafkaProduce会在批次填满或linger.ms达到上限时把批次发送出去。默认情况下，只要有用的线程，生产者就会把消息发送出去，就算批次里只有一个消息。把linger.ms设置成比0大的数，让生产者在发送批次之前等待一会儿，使更多的消息加入到这个批次。虽然这样会增加延迟，但会提升吞吐量（因为一次性发送更多的消息，每个消息的开销就变小了）

7.max.in.flight.requests.per.connection

该参数指定了生产者在到服务器晌应之前可以发送多少个消息。它的值越高，就会占用越多的内存，不过也会提升吞吐量。它设为1可以保证消息是按照发送的顺序写入服务器的，即使发生了重试。

思考：如何保证消息的有序性

Kafka 可以保证同一个分区里的消息是有序的。也就是说，如果生产者按照一定的顺序发送消息， broker 就会按照这个顺序把它们写入分区，消费者也会按照同样的顺序读取它们。在某些情况下，顺序是非常重要的。例如，往一个账户存入 100元再取出来，这个与先取钱再存钱是截然不同的！不过，有些场景对顺序不是很敏感。

如果把retries设为非零整数，同时把max.in.flight.requests.per.connection设为比1大的数，那么，如果第一个批次消息写入失败，而第二个批次写入成功， broker 会重试写入第一个批次。如果此时第一个批次也写入成功，那么两个批次的顺序就反过来了。

一般来说，如果某些场景要求消息是有序的，那么消息是否写入成功很关键的，所以不建议把retries设为0。可以把max.in.flight.requests.per.connection设为1，这样在生产者尝试发送第一批消息时，就不会有其他的消息发送给 broker 。不过这样会严重影响生产者的吞吐量，所以只有在对消息的顺序有严格要求的情况下才能这么做。