大数据-kafka学习（二）——生产者

痛定思痛。 2023-09-30 12:58 34阅读 0赞

## 一、发送流程 ##

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5py65pm65YW1_size_20_color_FFFFFF_t_70_g_se_x_16][]

## 二、生产者如何提高吞吐量 ##

> buffer.memory :RecordAccumulator 缓冲区大小，默认32m
> 
> batch.size: 缓冲区一批数据最大值，默认16k。适当增加该值，可以提高吞吐量，但是如果该值设置太大，会导致数据传输延迟增加。
> 
> Liger.ms ：如果数据迟迟未达到batch.size，sender等待linger.time之后就会发送数据，单位ms，默认值数0ms，表示没有延迟。生产环境建议该值大小为5-100ms之间
> 
> compression.type:生产者发送的所有数据压缩方式，默认数none，也就是不压缩。支持压缩类型：none、gzip、snappy、lz4和zstd。

## 三、数据可靠性 ##

### 1）acks： ###

> 0:生产者发送过来的数据，不需要等数据落盘后应答
> 
> 1:生产者发送过来的数据，Leader收到后应答
> 
> \-1:（all）：生产者发送过来的数据，Leader和Isa队列里
> 
> 面的所有节点收齐数据后应答，-1和all等价

### 2）至少一次： ###

> 至少一次（At Least Once）=ACK级别设置为-1 +分区副本大于等于2+ISR里应答的最小副本数量大于等于2

## 四、数据去重&事务 ##

### 1）配置参数 ###

> enable.idempotence 是否开启幂等性，默认true，表示开启幂等性。

### 2）Kafka的事务 ###

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

package my.kafka.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;
    
    /**
     * @author ：jizhibing
     * @date ：Created in 2022/3/31
     * @description：
     */
    public class MyKafkaProducerTransactions {
    
        private final static String KAFKA_SERVER_URL = "ip";
    
        public static void main(String[] args) throws Exception{
            //0 设置配置
            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
            //选择序列化方式
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,  StringSerializer.class.getName());
            //指定事务id
            properties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction_id_01") ;
    
            //1 ：建立连接
            KafkaProducer<String,String> kafkaProducer = new KafkaProducer(properties);
    
            //a\ 初始化事务
            kafkaProducer.initTransactions();
    
            //b、开始事务
            kafkaProducer.beginTransaction();
            try{
                //2：发送消息
                for (int i = 0; i < 100; i++) {
                    kafkaProducer.send(new ProducerRecord("first", "MY_transaction1"+i));
                }
                //制造失败
                int i = 1/0 ;
    
                //c、提交事务
                kafkaProducer.commitTransaction();
                System.out.println("success");
            }catch (Exception e){
                //d、回滚事务
                kafkaProducer.abortTransaction();
                System.out.println("fail");
            }finally {
                //3：关闭连接
                kafkaProducer.close();
            }
    
        }
    
    }

## 五、数据有序 ##

> 单分区内，有序（有条件的，不能乱序）
> 
> 多分区，分区与分区间无序；

## 六、生成经验-数据乱序 ##

> 1、kafka在1.x版本之前保证数据分区有序，条件如下：
> 
> **max.in.flight.requests.per.connection=1**(不需要考虑是否开启幂等性)。
> 
> 2、kafka在1.x及以后版本保证数据分区有序，条件如下：
> 
> （1）未开启幂等性
> 
> **max.in.flight.requests.per.connection**需要设置为1
> 
> （2）开启幂等性
> 
> **max.in.flight.requests.per.connection**需要设置小于等于5
> 
> 原因说明：因为在kafka 1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5哥request的数据都是有序的。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5py65pm65YW1_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5py65pm65YW1_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/8a6794fe698c449cb01561f2f8cf23e2.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5py65pm65YW1,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5py65pm65YW1_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/122c060c367d45298b75f5d19ef3de85.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5py65pm65YW1,size_20,color_FFFFFF,t_70,g_se,x_16