Kafka生产者

Kafka生产者概览

在这里插入图片描述

流程概述： 创建一个ProducerRecord对象，需要包含topic和要发送的内容，还可以指定键或分区。在发送ProducerRecord对象时，生产者要先把键和值对象序列化成字节数组。然后数据传给分区器，如果对象中指定了分区，那分区器直接把指定的分区返回，如果没有指定，分区器会根据ProducerRecord的键选择一个分区。这条记录随后被添加到记录批次里，这个批次里的所有消息会被发送到相同的主题和分区上。最后服务器收到消息成功写入会返回一个RecordMetaData对象，包含了主题和分区信息，以及记录在分区里的偏移量。如果写入失败，就会返回错误信息，生产者收到错误后会尝试重新发送，几次之后如果还是失败就返回错误信息。

创建Kafka生产者

这里只指定了必要的属性，其他使用默认配置

private Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);

指定了broker的地址清单，建议至少提供两个broker信息，其中一个宕机，生产者仍然能够连接到集群上。
指定了消息的键和值都是String

Kafka客户端默认提供了ByteArraySerializer、StringSerializer和IntegerSerializer接口的类。

发送消息到Kafka

简单配置

ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
try {
    producer.send(record);
} catch (Exception e) {
    e.printStackTrace();
}

这种方式不关心是否正常送达，虽然生产者会自动尝试重发，但不能完全保证消息不丢失。

同步发送消息

ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
try {
    producer.send(record).get();
} catch (Exception e) {
    e.printStackTrace();
}

send()方法会返回一个Future对象，调用get()方法进行等待可以知道消息是否发送成功。

异步发送消息

public class DemoProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null){
            exception.printStackTrace();
        }
    }
}

ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");
producer.send(record ,new DemoProducerCallback());

异步发送消息不等待响应，会提高发送速度，但如果遇到发送失败，可以使用回调进行一些处理。

顺序保证

Kafka可以保证同一个分区里的消息是有序的。如果把retries设为非零整数，把max.in.flight.requests.per.connection设为比1大的数。如果第一个批次消息写入失败，而第二个批次写入成功，broker会重试写入第一个批次，如果此时第一个批次也写入成功，那么两个批次的顺序就反过来了。如果某些场景要求消息是有序的，可以把max.in.flight.requests.per.connection设为1，这样在生产者尝试发送第一批消息时，不会有其他的消息发送给broker。不过这样做严重影响生产者的吞吐量。

分区

Kafka的消息是一个个键值对，ProducerRecord对象可以只包含目标主题和值，键可以设置为默认的null。键有两个用途：可以作为消息的附加信息，也可以用来决定消息该被写到topic的哪个分区。拥有相同键的消息将被写到同一个分区。

包含键值创建：

ProducerRecord<String, String> record = new ProducerRecord<>("topic11", "key111", "value111");

不包含键值创建：

ProducerRecord<String, String> record = new ProducerRecord<>("topic11",  "value111");

不包含键值的话，使用默认的分区器，记录将被随机地发送到topic各个可用的分区上。如果键不为空，使用了默认的分区器，那么Kafka会对键进行散列，然后根据散列值把消息映射到特定的分区上，同一个键总是被映射到同一个分区上，这也意味着，有可能会写入到不可用的分区而发生错误。
只有在不改变topic分区数量的情况下，键与分区的映射保持不变，一旦topic增加了新的分区，就无法保证了。下面实现自定义分区策略解决这一问题。

自定义分区策略

如下对target键做专一分区，其他键采用默认随机分配。

package com.qax.ngsoc.workorder.message.config;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.record.InvalidRecordException;
import org.apache.kafka.common.utils.Utils;

import java.util.List;
import java.util.Map;

public class MyPartitioner implements Partitioner {

    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
        int size = partitions.size();

        if (keyBytes == null || (!(key instanceof String))) {
            throw new InvalidRecordException("只接收字符串作为键");
        }

        if ("target".equals(key)) {
            return size;
        }

        return (Math.abs(Utils.murmur2(keyBytes)) % (size - 1));
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}