---

生产经验_生产者提高吞吐量

核心参数

Code

package com.artisan.pc;

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;

/**
 * @author 小工匠
 * @version 1.0
 * @mark: show me the code , change the world
 */
public class CustomProducerParameters {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.126.171: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());

        // batch.size：批次大小，默认16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        // linger.ms：等待时间，默认0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);

        // RecordAccumulator：缓冲区大小，默认32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // compression.type：压缩，默认none，可配置值gzip、snappy、lz4和zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("artisan", "art-msg-" + i));
        }

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

---

生产经验_数据可靠性

消息的发送流程

回顾下消息的发送流程如下：

ACK应答机制

背景	Kafka提供的解决方案
Leader收到数据，所有Follower开始同步数据，但有一个Follower因故障无法同步，导致Leader一直等待直到同步完成才发送ACK。	- Leader维护了一个动态的In-Sync Replica Set (ISR)和Leader保持同步的Follower集合。 - 当ISR中的Follower完成数据同步后，Leader向Producer发送ACK。 - 如果某个Follower长时间（replica.lag.time.max.ms）未向Leader同步数据，则该Follower将被移出ISR。 - 在Leader发生故障时，将从ISR中选举新的Leader。

---

ack应答级别

对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等ISR中的follower全部接收成功。
所以Kafka为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡，选择以下的配置

acks	描述
0	提供最低延迟，Leader副本接收消息后返回ack，尚未写入磁盘。可能导致数据丢失，特别是在Leader发生故障时。
1	Leader副本将消息写入磁盘后返回ack，但如果Leader在Follower副本同步数据之前发生故障，可能会丢失数据。
-1	或者 (all) ，Leader和所有Follower副本都将消息写入磁盘后才返回ack。如果在Follower副本同步完成后，Leader副本在发送ack之前发生故障，可能会导致数据重复。

---

应答机制 小结

---

Code

package com.artisan.pc;

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;

/**
 * @author 小工匠
 * @version 1.0
 * @mark: show me the code , change the world
 */
public class CustomProducerAck {

    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.126.171: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());

        // 设置acks
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // 重试次数retries，默认是int最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 10; i++) {
            kafkaProducer.send(new ProducerRecord<>("artisan", "art-msg-ack" + i));
        }

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

生产经验_数据去重

数据传递语义

---

幂等性

幂等性原理

开启幂等性配置（默认开启）

在prudocer的配置对象中，添加参数enable.idempotence,参数值默认为true，设置为false就关闭了。

---

生产者事务

kafka事务原理

事务代码流程

// 1初始化事务
void initTransactions();
// 2开启事务
void beginTransaction() throws ProducerFencedException;
// 3在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                              String consumerGroupId) throws ProducerFencedException;
// 4提交事务
void commitTransaction() throws ProducerFencedException;
// 5放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

---

生产经验_数据有序

---

生产经验_数据乱序

---