一篇文章读懂Kafka消息队列

我会带着你远行 2021-11-26 13:28 373阅读 0赞

### Kafka 基础篇 ###

*   *  消息队列的概念
     *  使用消息队列的场景分析
     *   *  异步消息发送：
         *  系统间解耦合
     *  Kafka 架构
     *  Kafka集群安装
     *   *  准备工作
         *  安装Zookeeper集群确保Kafka集群的正常运行
         *  Kafka安装步骤
         *  启动服务
         *  测试
     *  Topic 和 日志
     *  生产者
     *  消费者
     *  Topic管理篇（DDL）
     *   *  创建Tocpic
         *  Topic详细信息
         *  删除Topic
         *  Topic列表
     *  Kafka API实战(JDK1.8+)
     *   *  快速入门
         *   *  Maven依赖
             *  引入log4j.properies
             *  在Windos配置主机名和IP映射关系
             *  生产者
             *  消费者
     *  读取数据偏移量控制
     *  指定消费分区
     *  Kafka发送/接收Object
     *  生产者幂等性
     *  生产者批量发送
     *  生产者事务
     *   *  只有生产者
         *  生产者&消费者
     *  SpringBoot整合Kafka

## 消息队列的概念 ##

可以用于系统间通讯的一个组件-middle ware（中间件），该组件可以用于做消息缓冲的中间件（持久化）解决一些 并发处理、数据库缓冲等实现对高并发的业务场景的削峰填谷。

## 使用消息队列的场景分析 ##

### 异步消息发送： ###

> 使用Kafka MQ功能实现模块间异步通信，把一些费时的操作交给额外的服务或者设备去执行，这样可以提升系统运行效率，加速连接释放的速度，例如：用户注册模块，在用户注册成功后，业务系统需要给用户发送一个通知短信，通知用户登录邮箱去激活刚注册的用户信息。这种业务场景如图所示，因为短信通知和邮件发送是一个比较耗时的操作，所以在这里没必要将短信和邮件发送作为注册模块的流程，使用Message Queue功能可以将改业务和主业务注册分离，这样可以缩短用户浏览器和服务建立的链接时间，同时也能满足发送短信和邮件的业务。

### 系统间解耦合 ###

> ①在某些高吞吐的业务场景下，可能会出现在某一个时间段系统负载写入的负载压力比较大，短时间有大量的数据需要持久化到数据库中，但是由于数据的持久化需要数据库提供服务，由于传统的数据库甚至一些NoSQL产品也不能很好的解决高并发写入，因为数据库除去要向用户提供链接之外，还需要对新来的数据做持久化，这就需要一定的时间才能将数据落地到磁盘。因此在高并发写入的场景，就需要用户集成Message Queue在数据库前作为缓冲队列。在队列的另一头只需要程序有条不紊的将数据写入到数据库即可，这就保证无论外界写入压力有多么大都可以借助于Message Queue缓解数据库的压力。

> ②Message Queue除了解决对数据缓冲的压力之外，还可以充当业务系统的中间件（Middleware）作为系统服务间解耦的组件存在，例如上图所示订单模块和库存模块中就可以使用Message Queue作为缓冲队列实现业务系统服务间的解耦，也就意味着即使服务在运行期间库存系统宕机也并不会影响订单系统的正常运行。

## Kafka 架构 ##

Kafka集群以Topic形式负责管理集群中的Record，每一个Record属于一个Topic。底层Kafka集群通过日志分区形式持久化Record。在Kafka集群中，Topic的每一个分区都一定会有1个Borker担当该分区的Leader，其他的Broker担当该分区的follower（取决于分区的副本因子）。一旦对应分区的Lead宕机，kafka集群会给当前的分区指定新的Borker作为该分区的Leader。分区的Leader的选举是通过Zookeeper一些特性实现的，这里就不在概述了。Leader负责对应分区的读写操作，Follower负责数据备份操作。  
![Kafka架构图][Kafka]

## Kafka集群安装 ##

### 准备工作 ###

准备三台主机名分别为CentOSA|CentOSB|CentOSC的Linux系统主机  
分别关闭防火墙、相互做主机名映射、校对物理时钟、安装配置JDK8

### 安装Zookeeper集群确保Kafka集群的正常运行 ###

tar -zxf zookeeper-3.4.6.tar.gz -C /usr/
    mkdir /root/zkdata
    
    #分别在三台机器执行以下命令
    echo 1 >> /root/zkdata/myid
    echo 2 >> /root/zkdata/myid
    echo 3 >> /root/zkdata/myid
    
    touch /usr/zookeeper-3.4.6/conf/zoo.cfg
    vim /usr/zookeeper-3.4.6/conf/zoo.cfg

zoo.cfg

tickTime=2000
    dataDir=/root/zkdata
    clientPort=2181
    initLimit=5
    syncLimit=2
    
    server.1=CentOSA:2887:3887
    server.2=CentOSB:2887:3887
    server.3=CentOSC:2887:3887

启动zookeeper|查看zookeeper当前状态

/usr/zookeeper-3.4.6/bin/zkServer.sh start zoo.cfg
    /usr/zookeeper-3.4.6/bin/zkServer.sh status zoo.cfg

### Kafka安装步骤 ###

*  下载Kafka服务安装包http://archive.apache.org/dist/kafka/2.2.0/kafka\_2.11-2.2.0.tgz

tar -zxf kafka_2.11-2.2.0.tgz -C /usr
    vim /usr/kafka_2.11-2.2.0/config/server.properties
    ############################# Server Basics #############################
    broker.id=[0|1|2]  #三台机器分别 0/1/2
    ############################# Socket Server Settings #############################
    listeners=PLAINTEXT://CentOS[A|B|C]:9092 #三台机器分别A、B、C
    ############################# Log Basics #############################
    # A comma separated list of directories under which to store log files
    log.dirs=/usr/kafka-logs
    ############################# Zookeeper #############################
    zookeeper.connect=CentOSA:2181,CentOSB:2181,CentOSC:2181

> 注：此配置只能使用主机名访问 如需IP访问 将`listeners=PLAINTEXT://CentOS[A|B|C]:9092 #三台机器分别A、B、C`  
> 改为  
> `advertised.listeners=PLAINTEXT://x.x.x.x:9092`

### 启动服务 ###

cd /usr/kafka_2.11-2.2.0/
    ./bin/kafka-server-start.sh -daemon config/server.properties

### 测试 ###

*  **创建topic**

./bin/kafka-topics.sh --zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181 --create --topic topic01 --partitions 3 --replication-factor 3

*  **消费者**

./bin/kafka-console-consumer.sh  --bootstrap-server CentOSA:9092,CentOSB:9092,CentOSC:9092 --topic topic01

*  **生产者**

./bin/kafka-console-producer.sh --broker-list CentOSA:9092,CentOSB:9092,CentOSC:9092 --topic topic01

## Topic 和 日志 ##

Kafka集群是通过日志形式存储Topic中的Record，Record会根据分区策略计算得到的分区数存储到相应分区的文件中。每个分区都是一个有序的，不可变的记录序列，不断附加到结构化的commit-log中。每个分区文件会为Record进去分区的顺序进行编排。每一个分区中的Record都有一个id，该id标示了该record进入分区的先后顺序，通常将该id称为record在分区中的offset偏移量从0开始，依次递增。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70]

*  Kafka集群持久地保留所有已发布的记录 - 无论它们是否已被消耗 - 使用可配置的保留时间。例如，如果保留策略设置为2天，则在发布记录后的2天内，它可供使用，之后将被丢弃以释放空间。Kafka的性能在数据大小方面实际上是恒定的，因此长时间存储数据不是问题。
 *  事实上，基于每个消费者保留的唯一元数据是该消费者在日志中的偏移或位置。这种offset由消费者控制：通常消费者在读取记录时会线性地增加其偏移量，但事实上，由于消费者控制位置，它可以按照自己喜欢的任何顺序消费记录。例如，消费者可以重置为较旧的偏移量以重新处理过去的数据，或者跳到最近的记录并从“现在”开始消费。

## 生产者 ##

生产者负责发送Record到Kafka集群中的Topic中。在发布消息的时候，首先先计算Record分区计算方案有三种：

> ①如果用户没有指定分区但是指定了key信息，生产者会根据`hash（key）%分区数`计算该Record所属分区信息。  
> ②如果生产者在发送消息的时候并没有key，也没有指定分区数，生产者会使用**轮询策略**选择分区信息。  
> ③如果指定了分区信息，就按照指定的分区信息选择对应的分区；当分区参数确定以后生产者会找到相应分区的Leader节点将Record记录写入到Topic日志存储分区中。

## 消费者 ##

消费者作为消息的消费放，消费者对Topic中消息的消费方式是以**Group**为单位进行消费，Kafka服务器会自动的按照组内和组间对消费者消费的分区进行协调。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 1]

*  组内`均分分区`，确保一个组内的消费者不可重复消费分区中的数据，一般来说一个组内的消费者实例对的数目应该小于或者等于分区数目。
 *  组间`广播形式`消费，确保所有组都可以拿到当前Record。组间数据之间可以保证对数据的独立消费。

## Topic管理篇（DDL） ##

### 创建Tocpic ###

./bin/kafka-topics.sh
    --zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181 --create --topic topic01 --partitions 3 --replication-factor 3

### Topic详细信息 ###

./bin/kafka-topics.sh  --describe  --zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181  --topic topic01

### 删除Topic ###

./bin/kafka-topics.sh 
    --zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181  --delete  --topic topic01

> 如果用户没有配置`delete.topic.enable=true`，则Topic删除不起作用。

### Topic列表 ###

./bin/kafka-topics.sh  --zookeeper CentOSA:2181,CentOSB:2181,CentOSC:2181  --list

## Kafka API实战(JDK1.8+) ##

### 快速入门 ###

#### Maven依赖 ####

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.2.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-api</artifactId>
        <version>1.7.25</version>
    </dependency>
    <dependency>
        <groupId>log4j</groupId>
        <artifactId>log4j</artifactId>
        <version>1.2.17</version>
    </dependency>
    
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-log4j12</artifactId>
        <version>1.7.5</version>
    </dependency>

#### 引入log4j.properies ####

### set log levels ###
    log4j.rootLogger = info,stdout 
    ### 输出到控制台 ###
    log4j.appender.stdout = org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.Target = System.out
    log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern =%p %d %c %m %n

#### 在Windos配置主机名和IP映射关系 ####

192.168.111.128 CentOSA
    192.168.111.129 CentOSB
    192.168.111.130 CentOSC

#### 生产者 ####

package com.msk.demo01;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.text.DecimalFormat;
    import java.util.Properties;
    
    public class KafkaProducerDemo { 
        public static void main(String[] args) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
    
            //2.创建Kafka生产者
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
    
            //3.构建ProducerRecord
            for (int i=0;i<10;i++){ 
                DecimalFormat decimalFormat = new DecimalFormat("000");
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic04", decimalFormat.format(i), "value" + i);
                //4.发送消息
                producer.send(record);
            }
            //5.清空缓冲区
            producer.flush();
            //6.关闭生产者
            producer.close();
        }
    }

#### 消费者 ####

package com.msk.demo01;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Arrays;
    import java.util.Properties;
    
    public class KafkaConsumerDemo { 
        public static void main(String[] args) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
    
    
            //2.创建Kafka消费者
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
            //3.订阅topics
            consumer.subscribe(Arrays.asList("topic01"));
            //4.死循环读取消息
            while(true){ 
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if(records!=null && !records.isEmpty()){ 
                    for (ConsumerRecord<String, String> record : records) { 
                        int partition = record.partition();
                        long offset = record.offset();
                        long timestamp = record.timestamp();
                        String key = record.key();
                        String value = record.value();
                        System.out.println(partition+"\t"+offset+"\t"+timestamp+"\t"+key+"\t"+value);
                    }
                }
            }
        }
    }

## 读取数据偏移量控制 ##

默认当用户使用subscribe方式订阅topic消息， 默认首次offset策略是latest。当用户第一次订阅topic在消费者订阅之前的数据是无法消费到 消息的。用户可以配置消费端参数`auto.offset.reset`控制kafka消费者行为。

Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
    props.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
    
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");//默认值 latest

因为消费端在使用`consumer.poll`数据的时候，底层会定时的向Kafka服务器提交消费的偏移量。默认消费端的offset是自动提交的，用户如果不希望自动提交偏移量可以配置如下参数

> 注意如果用户使用subscribe方式订阅topic，在消费端必须指定`group.id`，这样Kafka才能够实现消费>端负载均衡以及实现组内均分组件广播。（推荐方式）

**默认配置**

enable.auto.commit	= true
    auto.commit.interval.ms	= 5000

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");

**手动提交偏移量**

public class KafkaConsumerDemo { 
        public static void main(String[] args) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
    
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);
            //2.创建Kafka消费者
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
            //3.订阅topics
            consumer.subscribe(Arrays.asList("topic01"));
            //4.死循环读取消息
            while(true){ 
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if(records!=null && !records.isEmpty()){ 
                    Map<TopicPartition, OffsetAndMetadata> offsetMeta=new HashMap<>();
                    for (ConsumerRecord<String, String> record : records) { 
                        int partition = record.partition();
                        long offset = record.offset();
                        long timestamp = record.timestamp();
                        String key = record.key();
                        String value = record.value();
                        System.out.println(partition+"\t"+offset+"\t"+timestamp+"\t"+key+"\t"+value);
    
                        TopicPartition part = new TopicPartition("topic03", partition);
                        OffsetAndMetadata oam=new OffsetAndMetadata(offset+1);//设置下一次读取起始位置
                        offsetMeta.put(part,oam);
                    }
                    consumer.commitSync(offsetMeta);
                }
            }
        }
    }

## 指定消费分区 ##

> 通过assign方式kafka对消费者的组管理策略失效。也就是说用户可以无需配置组ID。

public class KafkaConsumerDemo { 
        public static void main(String[] args) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            
            //2.创建Kafka消费者
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
            //3.指定分区
            consumer.assign(Arrays.asList(new TopicPartition("topic01",1)));
            consumer.seek(new TopicPartition("topic01",1),1);
            //4.死循环读取消息
            while(true){ 
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if(records!=null && !records.isEmpty()){ 
                    for (ConsumerRecord<String, String> record : records) { 
                        int partition = record.partition();
                        long offset = record.offset();
                        long timestamp = record.timestamp();
                        String key = record.key();
                        String value = record.value();
                        System.out.println(partition+"\t"+offset+"\t"+timestamp+"\t"+key+"\t"+value);
                    }
                }
            }
        }
    }

## Kafka发送/接收Object ##

**生产Object**

public interface Serializer<T> extends Closeable { 
       
        void configure(Map<String, ?> configs, boolean isKey);
        //重点实现serialize
        byte[] serialize(String topic, T data);
        default byte[] serialize(String topic, Headers headers, T data) { 
            return serialize(topic, data);
        }
        @Override
        void close();
    }

**消费Object**

public interface Deserializer<T> extends Closeable { 
    
        void configure(Map<String, ?> configs, boolean isKey);
        //重点实现方法
        T deserialize(String topic, byte[] data);
        default T deserialize(String topic, Headers headers, byte[] data) { 
            return deserialize(topic, data);
        }
        @Override
        void close();
    }

**实现序列化和反序列化**

public class ObjectCodec implements Deserializer<Object>, Serializer<Object> { 
        @Override
        public void configure(Map<String, ?> configs, boolean isKey) { 
            
        }
    
        @Override
        public byte[] serialize(String topic, Object data) { 
            return SerializationUtils.serialize((Serializable) data);
        }
    
        @Override
        public Object deserialize(String topic, byte[] data) { 
            return SerializationUtils.deserialize(data);
        }
    
        @Override
        public void close() { 
    
        }
    }

## 生产者幂等性 ##

> 幂等:多次操作最终的影响等价与一次操作称为幂等性操作,所有的读操作一定是幂等的.所有的写操作一定不是幂等的.当 生产者和broker默认有acks应答机制,如果当生产者发送完数据给broker之后如果没有在规定的时间内收到应答,生产者可以考虑重发数据.可以通过一下配置参数提升生产者的可靠性.

acks = all // 0 无需应答  n 应答个数 -1所有都需要
    retries = 3 // 表示重试次数
    request.timeout.ms = 3000 //等待应答超时时间
    enable.idempotence = true //开启幂等性

public class KafkaProducerDemo { 
        public static void main(String[] args) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
    
            props.put(ProducerConfig.ACKS_CONFIG,"all");//等待所有从机应答
            props.put(ProducerConfig.RETRIES_CONFIG,3);//重试3次
            props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);//等待3s应答
            props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);//开启幂等性
    
            //2.创建Kafka生产者
            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
    
            //3.构建ProducerRecord
            for (int i=15;i<20;i++){ 
                DecimalFormat decimalFormat = new DecimalFormat("000");
                User user = new User(i, "name" + i, i % 2 == 0);
                ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic01", decimalFormat.format(i), "user"+i);
                //4.发送消息
                producer.send(record);
            }
            //5.清空缓冲区
            producer.flush();
            //6.关闭生产者
            producer.close();
        }

## 生产者批量发送 ##

> 生产者会尝试缓冲record，实现批量发送，通过一下配置控制发送时机，记住如果开启可batch，一定在关闭producer之前需要flush。

batch.size = 16384 //16KB 缓冲16kb数据本地
    linger.ms = 2000 //默认逗留时间

public static void main(String[] args) { 
        //1.配置生产者了连接属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
    
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
    
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);//1kb缓冲区
        props.put(ProducerConfig.LINGER_MS_CONFIG,1000);//设置逗留时常
    
    
        //2.创建Kafka生产者
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
    
        //3.构建ProducerRecord
        for (int i=15;i<20;i++){ 
            DecimalFormat decimalFormat = new DecimalFormat("000");
            User user = new User(i, "name" + i, i % 2 == 0);
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic01", decimalFormat.format(i), "user"+i);
            //4.发送消息
            producer.send(record);
        }
        //5.清空缓冲区
        producer.flush();
        //6.关闭生产者
        producer.close();
    }

## 生产者事务 ##

> kafka生产者事务指的是在发送多个数据的时候，保证多个Record记录发送的原子性。如果有一条发送失败就回退，但是需要注意在使用kafka事务的时候需要调整消费者的事务隔离级别设置为`read_committed`因为kafka默认的事务隔离策略是`read_uncommitted`

开启事务

transactional.id=transaction-1 //必须保证唯一
    enable.idempotence=true //开启kafka的幂等性

### 只有生产者 ###

public class KafkaProducerDemo { 
        public static void main(String[] args) { 
    
            //1.创建Kafka生产者
            KafkaProducer<String, String> producer = buildKafkaProducer();
    
            //2.初始化事务和开启事务
            producer.initTransactions();
            producer.beginTransaction();
            try { 
                for (int i=5;i<10;i++){ 
                    DecimalFormat decimalFormat = new DecimalFormat("000");
                    User user = new User(i, "name" + i, i % 2 == 0);
                    ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic07", decimalFormat.format(i), "user"+i);
                    producer.send(record);
                }
                producer.flush();
                //3.提交事务]
                producer.commitTransaction();
            } catch (Exception e) { 
                System.err.println(e.getMessage());
                //终止事务
                producer.abortTransaction();
            }
            //5.关闭生产者
            producer.close();
        }
    
        private static KafkaProducer<String, String> buildKafkaProducer() { 
            //0.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
    
            props.put(ProducerConfig.ACKS_CONFIG,"all");
            props.put(ProducerConfig.RETRIES_CONFIG,3);
            props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);
            props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
    
            props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);//1kb缓冲区
            props.put(ProducerConfig.LINGER_MS_CONFIG,1000);//设置逗留时常
    
            //开启事务
            props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-"+UUID.randomUUID().toString());
            return new KafkaProducer<String, String>(props);
        }
    }

> 消费者那方需要将事务隔离级别设置为\`read\_committed

public class KafkaConsumerDemo { 
        public static void main(String[] args) { 
    
            //1.创建Kafka消费者
            KafkaConsumer<String, String> consumer = buildKafkaConsumer();
    
            //2.订阅topics
            consumer.subscribe(Arrays.asList("topic07"));
            //3.死循环读取消息
            while(true){ 
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                if(records!=null && !records.isEmpty()){ 
                    for (ConsumerRecord<String, String> record : records) { 
                        int partition = record.partition();
                        long offset = record.offset();
                        long timestamp = record.timestamp();
                        String key = record.key();
                        String value = record.value();
                        System.out.println(partition+"\t"+offset+"\t"+timestamp+"\t"+key+"\t"+value);
                    }
                }
            }
        }
    
        private static KafkaConsumer<String, String> buildKafkaConsumer() { 
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"CentOSA:9092,CentOSB:9092,CentOSC:9092");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            props.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");
            return new KafkaConsumer<String, String>(props);
        }
    }

### 生产者&消费者 ###

package com.msk.demo08;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.common.serialization.Deserializer;
    import org.apache.kafka.common.serialization.Serializer;
    
    import java.util.Properties;
    import java.util.UUID;
    
    public class KafkaUtils { 
        public static KafkaConsumer<String, String> buildKafkaConsumer(String servers, Class<? extends Deserializer> keyDeserializer,
                                                                       Class<? extends Deserializer> valueDeserializer,String group) { 
            Properties props = new Properties();
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializer);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializer);
            props.put(ConsumerConfig.GROUP_ID_CONFIG,group);
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed");
            props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);//设置为手动提交
            return new KafkaConsumer<String, String>(props);
        }
        public static KafkaProducer<String, String> buildKafkaProducer(String servers, Class<? extends Serializer> keySerializer,
                                                                       Class<? extends Serializer> valueSerializer) { 
            //1.配置生产者了连接属性
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,servers);
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,keySerializer);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,valueSerializer);
    
            props.put(ProducerConfig.ACKS_CONFIG,"all");
            props.put(ProducerConfig.RETRIES_CONFIG,3);
            props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,3000);
            props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
    
            props.put(ProducerConfig.BATCH_SIZE_CONFIG,1024);//1kb缓冲区
            props.put(ProducerConfig.LINGER_MS_CONFIG,1000);//设置逗留时常
    
            //开启事务
            props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,"transaction-"+ UUID.randomUUID().toString());
            return new KafkaProducer<String, String>(props);
        }
    }

**KafkaProducerAndConsumer**

package com.msk.demo08;
    
    import com.msk.demo05.User;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.apache.kafka.clients.consumer.OffsetAndMetadata;
    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.TopicPartition;
    import org.apache.kafka.common.serialization.StringDeserializer;
    import org.apache.kafka.common.serialization.StringSerializer;
    
    import java.text.DecimalFormat;
    import java.time.Duration;
    import java.util.*;
    
    public class KafkaProducerAndConsumer { 
        public static void main(String[] args) { 
    
            String servers = "CentOSA:9092,CentOSB:9092,CentOSC:9092";
            String group="g1";
            //1.创建Kafka生产者
            KafkaProducer<String, String> producer = KafkaUtils.buildKafkaProducer(servers,
                    StringSerializer.class, StringSerializer.class);
            KafkaConsumer<String, String> consumer = KafkaUtils.buildKafkaConsumer(servers,
                    StringDeserializer.class, StringDeserializer.class,group);
    
            consumer.subscribe(Arrays.asList("topic08"));
            //初始化事务
            producer.initTransactions();
    
            while (true) { 
                producer.beginTransaction();
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                try { 
                    Map<TopicPartition, OffsetAndMetadata> commits = new HashMap<TopicPartition, OffsetAndMetadata>();
                    for (ConsumerRecord<String, String> record : records) { 
                        TopicPartition partition = new TopicPartition(record.topic(), record.partition());
                        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1);
                        commits.put(partition, offsetAndMetadata);
    
                        System.out.println(record);
    
                        ProducerRecord<String, String> srecord = new ProducerRecord<String, String>("topic09", record.key(), record.value());
                        producer.send(srecord);
                    }
                    producer.flush();
    
                    //并没使用 consumer提交，而是使用producer帮助消费者提交偏移量
                    producer.sendOffsetsToTransaction(commits,group);
                    //提交生产者的偏移量
                    producer.commitTransaction();
                } catch (Exception e) { 
                    //System.err.println(e.getMessage());
                    producer.abortTransaction();
                }
            }
        }
    }

## SpringBoot整合Kafka ##

*  pom.xml

<properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <java.version>1.8</java.version>
        <kafka.version>2.2.0</kafka.version>
    </properties>
    
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.5.RELEASE</version>
    </parent>
    
    <dependencies>
    
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
    
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.5.RELEASE</version>
        </dependency>
        
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>
    </dependencies>

*  application.properties

server.port=8888
    
    # 生产者
    spring.kafka.producer.bootstrap-servers=CentOSA:9092,CentOSB:9092,CentOSC:9092
    spring.kafka.producer.acks=all
    spring.kafka.producer.retries=1
    spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
    spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
    
    # 消费者
    spring.kafka.consumer.bootstrap-servers=CentOSA:9092,CentOSB:9092,CentOSC:9092
    spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
    spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

*  代码

@SpringBootApplication
    @EnableScheduling
    public class KafkaApplicationDemo { 
        @Autowired
        private KafkaTemplate kafkaTemplate;
    
        public static void main(String[] args) { 
            SpringApplication.run(KafkaApplicationDemo.class,args);
        }
        @Scheduled(cron = "0/1 * * * * ?")
        public void send(){ 
            String[] message=new String[]{ "this is a demo","hello world","hello boy"};
            ListenableFuture future = kafkaTemplate.send("topic07", message[new Random().nextInt(message.length)]);
            future.addCallback(o -> System.out.println("send-消息发送成功：" + message), throwable -> System.out.println("消息发送失败：" + message));
        }
    
        @KafkaListener(topics = "topic07",id="g1")
        public void processMessage(ConsumerRecord<?, ?> record) { 
            System.out.println("record:"+record);
        }
    }

[Kafka]: /images/20211126/dc8b291f01944e10badaf2535a522b42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70]: /images/20211126/9c2faddbc07a44bcb8db6da3c4c5ea94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L00yODM1OTIzMzg_size_16_color_FFFFFF_t_70 1]: /images/20211126/9844a8247d2946dd98b586fe335b5840.png