03-kafka生产者

1、kafka生产者消息发送流程

1. 1 kafka发送原理

在消息发送的过程中，涉及到了两个线程——main 线程和Sender 线程。在main 线程中创建了一个双端队列RecordAccumulator。main 线程将消息发送给RecordAccumulator，Sender 线程不断从RecordAccumulator 中拉取消息发送到Kafka Broker。

kafka生产者消息发送流程

1.2 kafka生产者重要参数列表

参数名称	描述
bootstrap.servers	生产者连接集群所需的broker 地址清单。例如hadoop102:9092,hadoop103:9092,hadoop104:9092，可以设置1 个或者多个，中间用逗号隔开。注意这里并非需要所有的broker 地址，因为生产者从给定的 broker里查找到其他 broker 信息。
key.serializer和 value.serializer	指定发送消息的key 和 value 的序列化类型。一定要写全类名。
buffer.memory	RecordAccumulator缓冲区总大小，默认 32 m 。
batch.size	缓冲区一批数据最大值，默认 16 k 。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
linger.ms	如果数据迟迟未达到batch.size sender 等待 linger.time之后就会发送数据。单位 ms 默认值是 0ms ，表示没有延迟。生产环境建议该值大小为 5 -100 m s 之间。
acks	0：生产者发送过来的数据，不需要等数据落盘应答。1：生产者发送过来的数据 Leader 收到数据后应答。-1 all ）：生产者发送过来的数据 Leader+ 和 isr 队列里面的所有节点收齐数据后应答。默认值是 1, -1 和all 是等价的。
max.in.flight.requests.per.connection	允许最多没有返回ack 的次数，默认为 5 ，开启幂等性要保证该值是 1 5 的数字。
retries	当消息发送出现错误的时候，系统会重发消息。retries表示重试次数。默认是 int 最大值， 2147483647 。如果设置了重试，还想保证消息的有序性，需要设置MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION=1否则在重试此失败消息的时候，其他的消息可能发送成功了。
retry.backoff.ms	两次重试之间的时间间隔，默认是100 m s 。
enable.idempotence	是否开启幂等性默认 true ，开启幂等性
compression.type	生产者发送的所有数据的压缩方式。默认是 none ，也就是不压缩。支持压缩类型 none 、 gzip 、 snappy 、 lz4 和 zstd 。

2、异步发送API

添加kafka client

   <dependency>
         <groupId>org.apache.kafka</groupId>
         <artifactId>kafka-clients</artifactId>
         <version>3.0.0</version>
   </dependency>

2.1 普通异步发送

package com.xfm.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class AsyncProducer {

    public static void main(String[] args) {
        // 1. 创建kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "1.12.235.28:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG ,StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4.调用send方法发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("test02", "xfm" + i));
        }
        // 5.关闭资源
        kafkaProducer.close();
    }
}

2.2 带回调的异步发送

package com.xfm.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class AsyncProducerCallBack {
    public static void main(String[] args) {
        // 1. 创建kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "1.12.235.28:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG ,StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4.调用send方法发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("test02", "xfm" + i), new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("topic: " + recordMetadata.topic() + "   offset: " + recordMetadata.offset() + " partition: " + recordMetadata.partition() );
                    }
                }
            });
        }
        // 5.关闭资源
        kafkaProducer.close();
    }
}

返回结果

topic: test02   offset: 0 partition: 1
topic: test02   offset: 1 partition: 1
topic: test02   offset: 2 partition: 1
topic: test02   offset: 3 partition: 1
topic: test02   offset: 4 partition: 1
topic: test02   offset: 5 partition: 1
topic: test02   offset: 6 partition: 1
topic: test02   offset: 7 partition: 1
topic: test02   offset: 8 partition: 1
topic: test02   offset: 9 partition: 1

3、同步发送API

同步发送和异步发送的区别只是在send()方法后调用get()方法；

// 异步发送
 kafkaProducer.send(new ProducerRecord<String, String>("test02", "xfm" + i));
// 同步发送
 kafkaProducer.send(new ProducerRecord<String, String>("test02", "xfm" + i)).get();

4、生产者分区

4.1 分区的好处

（1）便于合理使用储存资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一
块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
（2）提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据。

4.2 生产者发送消息的分区策略

（1）默认的分区器DefaultPartitioner

首先看看源码

/**
 * The default partitioning strategy:
 * <ul>
 * <li>If a partition is specified in the record, use it
 * <li>If no partition is specified but a key is present choose a partition based on a hash of the key
 * <li>If no partition or key is present choose the sticky partition that changes when the batch is full.
 * See KIP-480 for details about sticky partitioning.
 */
public class DefaultPartitioner implements Partitioner {
...
}

从源码的注释中我们可以得知：
（1）指明partition的情况下，直接将指明的值作为partition值；例如partition=0，所有数据写入
分区0
（2）没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区
（3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）。

案例一

将数据发往指定partition 的情况下，例如，将所有数据发往分区1 中。

package com.xfm.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) {
        // 1. 创建kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "1.12.235.28:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG ,StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4.调用send方法发送消息
        for (int i = 0; i < 10; i++) {
            // 指定4号分区，key为空
            kafkaProducer.send(new ProducerRecord<String, String>("test02",4, "", "xfm" + i));
        }
        // 5.关闭资源
        kafkaProducer.close();
    }
}

结果：

topic: test02   offset: 10 partition: 4
topic: test02   offset: 11 partition: 4
topic: test02   offset: 12 partition: 4
topic: test02   offset: 13 partition: 4
topic: test02   offset: 14 partition: 4
topic: test02   offset: 15 partition: 4
topic: test02   offset: 16 partition: 4
topic: test02   offset: 17 partition: 4
topic: test02   offset: 18 partition: 4
topic: test02   offset: 19 partition: 4

案例二

没有指明
partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值。

package com.xfm.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class CustomProducerCallbackPartitions {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "1.12.235.28:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG , StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG ,StringSerializer.class.getName());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        String[] keys = new String[]{"a", "b", "c"};
        // 4.调用send方法发送消息
        for (int j = 0; j < keys.length; j++) {
            System.out.println("当前的key值：" + keys[j]);
            for (int i = 0; i < 10; i++) {
                kafkaProducer.send(new ProducerRecord<String, String>("test02", keys[j], "xfm" + i), new Callback() {
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        if (e == null) {
                            System.out.println("topic: " + recordMetadata.topic() + "   offset: " + recordMetadata.offset() + " partition: " + recordMetadata.partition() );
                        }
                    }
                }).get();
            }
        }

        // 5.关闭资源
        kafkaProducer.close();
    }
}

结果

当前的key值：a
topic: test02   offset: 40 partition: 4
topic: test02   offset: 41 partition: 4
topic: test02   offset: 42 partition: 4
topic: test02   offset: 43 partition: 4
topic: test02   offset: 44 partition: 4
topic: test02   offset: 45 partition: 4
topic: test02   offset: 46 partition: 4
topic: test02   offset: 47 partition: 4
topic: test02   offset: 48 partition: 4
topic: test02   offset: 49 partition: 4
当前的key值：b
topic: test02   offset: 20 partition: 1
topic: test02   offset: 21 partition: 1
topic: test02   offset: 22 partition: 1
topic: test02   offset: 23 partition: 1
topic: test02   offset: 24 partition: 1
topic: test02   offset: 25 partition: 1
topic: test02   offset: 26 partition: 1
topic: test02   offset: 27 partition: 1
topic: test02   offset: 28 partition: 1
topic: test02   offset: 29 partition: 1
当前的key值：c
topic: test02   offset: 50 partition: 4
topic: test02   offset: 51 partition: 4
topic: test02   offset: 52 partition: 4
topic: test02   offset: 53 partition: 4
topic: test02   offset: 54 partition: 4
topic: test02   offset: 55 partition: 4
topic: test02   offset: 56 partition: 4
topic: test02   offset: 57 partition: 4
topic: test02   offset: 58 partition: 4
topic: test02   offset: 59 partition: 4

4.3 自定义分区器

研发人员可以根据企业需求，自己重新实现分区器

1）需求

例如我们实现一个分区器实现，
发送过来的数据中如果包含 xfm ，就发往 0 号分区，不包含 xfm ，就发往 1 号分区；

2）实现步骤

（1）定义类实现Partitioner接口
（2）重写partition()方法

package com.xfm.producer.partition;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;

/**
 * 自定义分区器
 */
public class MyPartitioner implements Partitioner {
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        // 获取消息
        String msgValue = value.toString();
        // 创建 partition
        int partition;
        // 判断消息是否包含 xfm
        if(msgValue.contains("xfm")){
            partition = 0;
        } else {
            partition = 1;
        }
        // 返回分区号
        return partition;
    }

    public void close() {

    }

    public void configure(Map<String, ?> map) {

    }
}

（3）使用分区器的方法，在生产者的配置中添加分区器参数。

package com.xfm.producer;

import com.xfm.producer.partition.MyPartitioner;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class ProducerCallbackCustomPartitioner {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "1.12.235.28:9092");
        // key,value 序列化（必须）：key.serializer，value.serializer
        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());
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4.调用send方法发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<String, String>("test02", "xfm" + i), new Callback() {
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    if (e == null) {
                        System.out.println("topic: " + recordMetadata.topic() + "   offset: " + recordMetadata.offset() + " partition: " + recordMetadata.partition() );
                    }
                }
            }).get();
        }
        

        // 5.关闭资源
        kafkaProducer.close();
    }
}

结果

topic: test02   offset: 10 partition: 0
topic: test02   offset: 11 partition: 0
topic: test02   offset: 12 partition: 0
topic: test02   offset: 13 partition: 0
topic: test02   offset: 14 partition: 0
topic: test02   offset: 15 partition: 0
topic: test02   offset: 16 partition: 0
topic: test02   offset: 17 partition: 0
topic: test02   offset: 18 partition: 0
topic: test02   offset: 19 partition: 0

5、生产经验-生产者如何提高吞吐量

主要思路是一批次传输的数据量大点，缓冲区大小调大，对传输的数据进行压缩
（1）batch.size: 批次大小，默认16k
（2）linger.ms：等待时间，修改为5-100ms
（3） compression.type：压缩snappy
（4）RecordAccumulator：缓冲区大小，修改为64m

生产者提高吞吐量

6、生产经验-如何保证数据可靠性

首先回顾一下生产者发送流程（本章第一节kafka发送原理）

1）ack应答原理

ack=0: 生产者发送过来的数据，不需要等数据落盘应答

ack=0时
ack=1:生产者发送过来的数据，Leader收到数据后应答

ack=1时
ack=-1: 生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答

akc=-1时

思考：Leader收到数据，所有Follower都开始同步数据，但有一个Follower，因为某种故障，迟迟不能与Leader进行同步，那这个问题怎么解决呢？

Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合(leader：0，isr:0,1,2)。如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如2超时，(leader:0, isr:0,1)。
这样就不用等长期联系不上或者已经故障的节点。

数据可靠性分析：
如果分区副本设置为1个，或者ISR里应答的最小副本数量（ min.insync.replicas 默认为1）设置为1，和ack=1的效果是一样的，仍然有丢数的风险（leader：0，isr:0）。

数据完全可靠条件= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

可靠性总结：
acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

数据重复分析：
acks：-1（ all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答。

数据重复分析

7、生产经验-数据去重

7.1 数据传递语义

至少一次（At Least Once）=ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2
最多一次（AtMost Once）= ACK级别设置为0
总结：
At Least Once可以保证数据不丢失，但是不能保证数据不重复；
At Most Once可以保证数据不重复，但是不能保证数据不丢失。
精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。
Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

7.2 幂等性

幂等性原理

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条，保证了不重复。
精确一次（Exactly Once） = 幂等性+ 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2）

重复数据的判断标准：

具有<PID, Partition, SeqNumber>相同主键的消息提交时，Broker只会持久化一条。其
中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的。
所以幂等性只能保证的是在单分区单会话内不重复。

幂等性原理

如何使用幂等性

开启参数enable.idempotence 默认为true，false 关闭。

7.3 生产者事务

1）kafka事务原理

说明：开启事务，必须开启幂等性。
Producer 在使用事务功能前，必须先自定义一个唯一的transactional.id。有了transactional.id，即使客户端挂掉了，它重启后也能继续处理未完成的事务
由于篇幅原因，详情请查看https://www.jianshu.com/p/057dc195428d

最后编辑于：2022.05.15 23:23:32

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 219,366评论 6赞 508
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 93,521评论 3赞 395
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 165,689评论 0赞 356
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 58,925评论 1赞 295
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 67,942评论 6赞 392
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 51,727评论 1赞 305
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,447评论 3赞 420
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,349评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,820评论 1赞 317
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,990评论 3赞 337
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 40,127评论 1赞 351
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,812评论 5赞 346
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,471评论 3赞 331
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 32,017评论 0赞 22
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,142评论 1赞 272
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,388评论 3赞 373
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 45,066评论 2赞 355

03-kafka生产者

1、kafka生产者消息发送流程

1. 1 kafka发送原理

1.2 kafka生产者重要参数列表

2、异步发送API

2.1 普通异步发送

3、同步发送API

4、生产者分区

4.1 分区的好处

4.2 生产者发送消息的分区策略

（1）默认的分区器DefaultPartitioner

案例一

案例二

4.3 自定义分区器

1）需求

2）实现步骤

5、 生产经验-生产者如何提高吞吐量

6、 生产经验-如何保证数据可靠性

1）ack应答原理

数据完全可靠条件= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

7、 生产经验-数据去重

7.1 数据传递语义

7.2 幂等性

幂等性原理

重复数据的判断标准：

如何使用幂等性

7.3 生产者事务

1）kafka事务原理

推荐阅读更多精彩内容

5、生产经验-生产者如何提高吞吐量

6、生产经验-如何保证数据可靠性

7、生产经验-数据去重