Kafka1.0.X_生产者API详解

Producer是Kafka三大组件中的一个，用于发送消息到kafka集群中

Producer提供了丰富的配置（见后面的配置项）用于控制它的行为

在编码以前先使用命令建立topic

./kafka-topics.sh --create --zookeeper hadoop01,hadoop02,hadoop03:2181 --partitions 2 --replication-factor 3 --topic mytopic1

简单模式

Producer采用默认分区方式将消息散列的发送到各个分区当中

package com.jv;
​
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
​
import java.util.Properties;
​
public class MyProducer {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //设置kafka集群的地址
        props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
        //ack模式，all是最慢但最安全的
        props.put("acks", "-1");
        //失败重试次数
        props.put("retries", 0);
        //每一个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
        props.put("batch.size", 10);
        //props.put("max.request.size",10);
        //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
        props.put("linger.ms", 10000);
        //整个Producer用到总内存的大小，若是缓冲区满了会提交数据到服务端
        //buffer.memory要大于batch.size，不然会报申请内存不足的错误
        props.put("buffer.memory", 10240);
        //序列化器
        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>("mytopic1", Integer.toString(i), "dd:"+i));
        //Thread.sleep(1000000);
        producer.close();
    }
}

配置项

名称	说明	默认值	有效值	重要性
bootstrap.servers	kafka集群的broker-list，如：<br>hadoop01:9092,hadoop02:9092	无		必选
acks	确保生产者可靠性设置，有三个选项：<br>acks=0:不等待成功返回<br>acks=1:等Leader写成功返回<br>acks=all:等Leader和全部ISR中的Follower写成功返回,all也能够用-1代替	-1	0,1,-1,all
key.serializer	key的序列化器		ByteArraySerializer<br>StringSerializer	必选
value.serializer	value的序列化器		ByteArraySerializer<br>StringSerializer	必选
buffer.memory	Producer整体内存大小	33554432	不要超过物理内存，根据实际状况调整	建议必选
compression.type	压缩类型<br>压缩最好用于批量处理，批量处理消息越多，压缩性能越好	无	none、gzip、snappy
retries	发送失败尝试重发次数	0
batch.size	每一个partition的未发送消息大小	16384	根据实际状况调整	建议必选
client.id	附着在每一个请求的后面，用于标识请求是从什么地方发送过来的
connections.max<br>.idle.ms	链接空闲时间超过太久自动关闭（单位毫秒）	540000
linger.ms	数据在缓冲区中保留的时长,0表示当即发送<br>为了减小网络耗时，须要设置这个值<br>太大可能容易致使缓冲区满，阻塞消费者<br>过小容易频繁请求服务端	0
max.block.ms	最大阻塞时长	60000
max.request.size	请求的最大字节数，该值要比batch.size大<br>不建议去更改这个值，若是设置很差会致使程序不报错，但消息又没有发送成功	1048576
partitioner.class	分区类，能够自定义分区类，实现partitioner接口	默认是哈希值%partitions
receive.buffer.bytes	socket的接收缓存空间大小,当阅读数据时使用	32768
request.timeout.ms	等待请求响应的最大时间,超时则重发请求,超太重试次数将抛异常	3000
send.buffer.bytes	发送数据时的缓存空间大小	131072
timeout.ms	控制server等待来自followers的确认的最大时间	30000
max.in.flight.<br>requests.per.<br>connection	kafka能够在一个connection中发送多个请求，叫做一个flight,这样能够减小开销，可是若是产生错误，可能会形成数据的发送顺序改变。	5
metadata.fetch<br>.timeout.ms	从ZK中获取元数据超时时间<br>好比topic\host\partitions	60000
metadata.max.age.ms	即便没有任何partition leader 改变，强制更新metadata的时间间隔	300000
metric.reporters	类的列表，用于衡量指标。实现MetricReporter接口，将容许增长一些类，这些类在新的衡量指标产生时就会改变。JmxReporter总会包含用于注册JMX统计	none
metrics.num.samples	用于维护metrics的样本数	2
metrics.sample.window.ms	metrics系统维护可配置的样本数量，在一个可修正的window size。这项配置配置了窗口大小，例如。咱们可能在30s的期间维护两个样本。当一个窗口推出后，咱们会擦除并重写最老的窗口	30000
reconnect.backoff.ms	链接失败时，当咱们从新链接时的等待时间。这避免了客户端反复重连	10
retry.backoff.ms	在试图重试失败的produce请求以前的等待时间。避免陷入发送-失败的死循环中	100

更全的配置参考官方文档：http://kafka.apache.org/documentation/#producerconfigs

幂等模式

​ 幂等性：客户端一次或屡次操做，最终数据是一致的，好比购买火车票支付时可能显示网络异常，但其实已经扣款成功，用户再次发起扣款不会再触发真正的扣款Kafka只能保证在一个会话中的幂等性

​ 幂等模式只须要将enable.idempotence设置为true，一旦设置了该属性，那么retries默认是Integer.MAX_VALUE ，acks默认是all。代码的写法和前面例子没什么区别

事务模式

​事务模式要求数据发送必须包含在事务中，在事务中能够向多个topic发送数据，消费者端最好也使用事务模式读，保证一次能将整个事务的数据所有读取过来。固然消费者也能够不设置为事务读的模式。

   @Test
    public void transactional(){
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
        props.put("transactional.id", "my_transactional_id");
        Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
​
        producer.initTransactions();
​
        try {
            //数据发送必须在beginTransaction()和commitTransaction()中间，不然会报状态不对的异常
            producer.beginTransaction();
            for (int i = 0; i < 100; i++)
                producer.send(new ProducerRecord<>("mytopic1", Integer.toString(i), Integer.toString(i)));
            producer.commitTransaction();
        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
            // 这些异常不能被恢复，所以必需要关闭并退出Producer
            producer.close();
        } catch (KafkaException e) {
            // 出现其它异常，终止事务
            producer.abortTransaction();
        }
        producer.close();
    }

自定义分区类(Partitioner)

没什么特殊分区逻辑，把kafka默认的分区类拿出来读一下

package org.apache.kafka.clients.producer.internals;
​
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.utils.Utils;
​
public class DefaultPartitioner implements Partitioner {
    private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap();
    //须要覆盖的方法
    public DefaultPartitioner() {
    }
    //须要覆盖的方法，能够在这里添加配置信息
    public void configure(Map<String, ?> configs) {
    }
    //须要覆盖的方法，最重要的
    /*
    topic:主题
    key:动态绑定的，传的什么类型就是什么类型
    keyBytes:Ascii码数组
    value:动态绑定的，传的什么类型就是什么类型
    valueBytes:Ascii码数组
    cluster:kafka集群
    */
    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();
        //若是key为空，则取消息做为分区依据
        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 {
                //若是可用分区=0，则直接返回全部分区中的一个
                return Utils.toPositive(nextValue) % numPartitions;
            }
        } else {
            //key有值，则返回全部分区中的一个
            return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
        }
    }
    //若是没有key，则调用该方法那消息来作分区依据
    private int nextValue(String topic) {
        AtomicInteger counter = (AtomicInteger)this.topicCounterMap.get(topic);
        if (null == counter) {
            counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
            AtomicInteger currentCounter = (AtomicInteger)this.topicCounterMap.putIfAbsent(topic, counter);
            if (currentCounter != null) {
                counter = currentCounter;
            }
        }
​
        return counter.getAndIncrement();
    }
    //须要覆盖的方法
    public void close() {
    }
}