1.Kafka讲解

3.1 什么是Kafka

kafka官网：http://kafka.apache.org/

中文文档：https://www.orchome.com/511

​ Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue）

3.2 Kafka的特性

1. 高吞吐量、低延迟

   kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, consumer group 对partition进行consume操作。

2. 可扩展性

   kafka集群支持热扩展

3. 持久性、可靠性

   消息被持久化到本地磁盘，并且支持数据备份防止数据丢失。默认最多保存时长 为7天

4. 容错性

   允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

5. 高并发

   支持数千个客户端同时读写

1.1 消息队列中间件概述

​ 消息队列中间件是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题实现高性能，高可用，可伸缩和最终一致性 .

​ 常见的消息队列产品有ActiveMQ，RabbitMQ， RocketMQ ,Kafka,ZeroMQ,MetaMQ

ActiveMQ： 很老了，已被市场淘汰了。

RabbitMQ：用于传统企业项目（erp,oa,crm…）。erlang, 速度快，消息存放在内存中

RocketMQ: 用于电商、金融。速度快，支持亿级别消息堆积，模拟了kafka。

Kafka: 主要用海量数据，高吞吐量。用于日志收到与传输上。

选择MQ优点适合自己项目的才是真的

2.Kafa的架构

1）Producer ：消息生产者，就是向 kafka broker 发消息的客户端；
2）Consumer ：消息消费者，向 kafka broker 取消息的客户端；
3）Consumer Group （CG）：消费者组，由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消者组是逻辑上的一个订阅者。可以提高消费者的能力（并发最好的时候： 消费者组里面消费者的个数=Topic里面的分区数）。
4）Broker ：一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker可以容纳多topic。
5）Topic ：可以理解为消息的分类，生产者和消费者面向的都是同一个 topic；
6）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；(往同一个主题发送的消息不会同时发给多个分区)，存放消息的地方

7）Replica（slave）：副本，为保证集群中的某个节点发生故障时，该节点上的 partition 数据不丢失，且kafka 仍然能够继续工作，kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 Master和若干个 Slave(Replica)。
8）leader[master]：每个分区(针对Patition)多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 leader。
9）follower[slave]：每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。leader 发生故障时，某个 follower 会成为新的 leader。

2.2.小结

生产者 -> kafka broker(1个topic有多个partition(消息存储))

消费者->kafka broker上的topic下的分区，一个分区只能被同一消费者组中的1个消费者消费

注册中心：zookeeper 注册 集群维护

3.kafka安装（也可以参考我博客【docker的安装与使用】）

(1) 容器下载，看博客资料

docker pull wurstmeister/zookeeper
docker pull wurstmeister/kafka
docker pull obsidiandynamics/kafdrop:3.30.0

docker run -d --name zookeeper --restart=always -p 2181:2181 wurstmeister/zookeeper

docker volume create kafka-log

docker run -d --name kafka -p 9092:9092 \
-e KAFKA_BROKER_ID=0 \
-e KAFKA_ZOOKEEPER_CONNECT=172.16.147.129:2181/kafka \
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.16.147.129:9092 \
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
-e KAFKA_HEAP_OPTS="-Xmx512M -Xms256M" \
-v kafka-log:/kafka \
wurstmeister/kafka

docker run -d -p 9100:9000 --name=kafdrop \
    -e KAFKA_BROKERCONNECT=172.16.147.129:9092 \
    -e JVM_OPTS="-Xms32M -Xmx64M" \
    -e SERVER_SERVLET_CONTEXTPATH="/" \
    obsidiandynamics/kafdrop:3.30.0
# 启动成功后访问
http://172.16.147.129:9100
————————————————
版权声明：本文为CSDN博主「添砖Java的亮哥」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/m0_51262858/article/details/127596051

# 参数说明
KAFKA_BROKER_ID: 分区id, 如果不指定，默认也是从0开始
KAFKA_ZOOKEEPER_CONNECT: zookeeper所在
KAFKA_ADVERTISED_LISTENERS: 客户端连接地址与端口 可对外公网
KAFKA_LISTENERS：内网使用 客户端连接地址与端口
KAFKA_HEAP_OPTS：使用堆大小配置

访问页面如下：

（2）官网下载
下载地址：http://kafka.apache.org/downloads
下载的安装包上传到公司服务器，解压
（3）修改参数

修改config目录下的server.properties文件,效果如下

• 修改listeners=PLAINTEXT://host:9092
• log.dirs=/root/kafka_2.12-2.2.1/logs
  bin/kafka-server-start.sh config/server.properties #启动kafka
  jsp查看进程

4.小结

注意:

• 安装Kafka之前先需要安装JDK

• 启动kafka之前，必须先启动zookeeper

4.Kafka的使用（原生方式与Spring Boot整合方式）原生早就不用了，太垃圾了

创建kafka-demo工程，引入依赖信息

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.8.RELEASE</version>
</parent>

<dependencies>
    <!-- kafka依赖 begin -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
    <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.8.3</version>
    </dependency>
</dependencies>

做一个java普通的生产者和消费者只需要依赖kafka-clients即可

3.2 消息生产者

创建类：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.jupiter.api.Test;

import java.util.Properties;

/**
 * @version 1.0
 * @description 说明
 * @package
 */
public class ProducerTest {

    @Test
    public void testSend(){
        //1. 创建配置信息
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.211.128:9092");
        // key的序列化器
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        // 内容的序列化器
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        // 重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG,10);
        //2. 创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
        //3. 创建消息对象
        // 构建参数1：主题， 2：消息内容
        ProducerRecord<String,String> msg = new ProducerRecord<String,String>("myTopic", "Hello kafka...");
        //4. 发送消息
        producer.send(msg);
        producer.close();
    }
}

3.3 消息消费者

创建消费者类：

 
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 org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
 
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
 
/**
 * @version 1.0
 * @description 说明
 * @package
 */
public class ConsumerDemo {
 
    private static final String TOPIC = "myTopic";
 
    public static void main(String[] args) {
        //1.创建配置对象
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.211.128:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");//设置分组
 
        //2.创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        //3.订阅主题
        consumer.subscribe(Collections.singletonList(TOPIC));
        while (true){
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record.value());
                System.out.println(record.key());
            }
        }
    }
}
 

4 测试及结论

• 配置文件看ProducerConfig, ConsumerConfig (有哪些配置项、配置项的名称、有哪些值)

• 生产者发送消息，同一个组中的多个消费者只能有一个消费者接收消息
• 生产者发送消息，如果有多个组，每个组中只能有一个消费者接收消息,如果想要实现广播的效果，可以让每个消费者单独有一个组即可，这样每个消费者都可以接收到消息

使用步骤：

1. 创建配置信息Properties：

   • 服务器ip与端口BOOTSTRAP_SERVER
   • 生产者（序列化器），消费者（反序列化器）

2. 创建生产者对象|或消费者对象

3. 生产者，创建消息（指定主题，[指定key], 消息内容），发送

4. 消费者订阅主主题（是一个集合，可以多个主题），拉取消息，消费

相关参数：

• batch.size：只有数据积累到 batch.size 之后，sender 才会发送数据。(默认 16384byte, 16KB)
• linger.ms：如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。(默认为0)

4.1 参数详解

​ 到目前为止，我们只介绍了生产者的几个必要参数（bootstrap.servers、序列化器等）

​ 生产者还有很多可配置的参数，在kafka官方文档中都有说明，大部分都有合理的默认值，所以没有必要去修改它们，不过有几个参数在内存使用，性能和可靠性方法对生产者有影响

http://kafka.apache.org/23/documentation.html#producerconfigs

https://www.orchome.com/511

4.2 消息重复消费处理

幂等：不管执行多少次，结果都一样

Get, Put, Delete, 幂等

1. 判断数据的状态, status=1 需要审核, 如果status!=1不处理
2. 只有一个消费者组，一个消息只会被一个消费者消费
3. 加乐观锁，取消息记录版本(100,15:56) (15:57,100)，更新数据时判断版本是否一致(15:56 100)，一致则更新，不一致则不更新数据, affected rows: 0代表没更新成功，其它消费者已经处理过了
4. 发送消息时，给消息唯一标识, 当消费消息时，先检查这个唯一的id是否存在，存在则不处理，不存在则存入redis|mysql，开始处理。

5.spring boot集成kafka收发消息，在项目中使用

1.生产者端和消费者端都要引入依赖

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.8.RELEASE</version>
</parent>

<dependencies>
    <!-- kafka依赖 begin -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

2.生产者端配置yml

spring:
  kafka:
    # 配置连接到服务端集群的配置项 ip:port,ip:port
    bootstrapServers: 172.16.147.129:9092
    producer:
      # 当多个消息要发送到相同分区的时，生产者尝试将消息批量打包在一起，以减少请求交互。这样有助于客户端和服务端的性能提升。该配置的默认批次大小（以字节为单位）
      batchSize: 16384
      # 生产者用来缓存等待发送到服务器的消息的内存总字节数。如果消息发送比可传递到服务器的快，生产者将阻塞max.block.ms之后，抛出异常
      bufferMemory: 33554432
      # 发送失败则会重新发送,间隔100ms
      retries: 0
      keySerializer: org.apache.kafka.common.serialization.StringSerializer
      valueSerializer: org.apache.kafka.common.serialization.StringSerializer

测试发送消息

/**
 * @author zengfl
 * @Date 上午12:05
 * @User 添砖Java的亮哥
 */
@SpringBootTest
public class SendMsgTest {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Test
    public void testSendMsg(){
    kafkaTemplate.send("Topic.liangge","hello !  kafka");
    }
}

查看

3.消费者消费配置

配置消费者的yml

spring:
  kafka:
    consumer:
      # autoCommitInterval: 100
      autoOffsetReset: earliest
      # 是否自动提交偏移量，默认为true自动提交，false为手工提交
      # enable-auto-commit: false
      groupId: test-consumer-group
      # 默认值即为字符串
      keyDeserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 默认值即为字符串
      valueDeserializer: org.apache.kafka.common.serialization.StringDeserializer
      # 配置连接到服务端集群的配置项 ip:port,ip:port
      bootstrap-servers: 172.16.147.129:9092

创建消费者的监听类，消费消息

/**
 * @author zengfl
 * @Date 上午12:24
 * @User 添砖Java的亮哥
 */
@Component
public class TestListener {
    @KafkaListener(topics = "Topic.liangge",groupId = "group_01")
    public void consumeMsg(String msg){
        System.out.println("111111==========>" + msg);
    }

    @KafkaListener(topics = "Topic.liangge",groupId = "group_02")
    public void consumeMsg2(ConsumerRecord<String, String> record){
        System.out.println("=================2222222==================");
        System.out.println(String.format("==> key=%sd value=%s offset=%d",record.key(),record.value(),record.offset()));
        System.out.println("==========================================");
    }
}

消费者可以看到监听到Kafka的消息

在试试
@SpringBootTest
public class SendMsgTest {

@Autowired
private KafkaTemplate kafkaTemplate;

@Test
public void testSendMsg(){
kafkaTemplate.send("Topic.liangge","愿疫情早日结束");
}

}

测试及结论

• 生产者发送消息，同一个组中的多个消费者只能有一个消费者接收消息
• 生产者发送消息，如果有多个组，每个组中只能有一个消费者接收消息,如果想要实现广播的效果，可以让每个消费者单独有一个组即可，这样每个消费者都可以接收到消息
• offset： 消息在队列中的下标，从0开始