Docker搭建kafka+zookeeper以及Springboot集成kafka快速入门

参考文章
【Docker安装摆设Kafka+Zookeeper具体教程】_linux arm docker安装kafka-CSDN博客

Docker搭建kafka+zookeeper


打开我们的docker的镜像源设置
   vim /etc/docker/daemon.json
  设置

    {
  "registry-mirrors": ["https://widlhm9p.mirror.aliyuncs.com"]
}
   下面的谁人insecure是我自己捏造机的，不消剖析

拉取镜像

然后开始拉取我们的zookeeper镜像和我们的kafka镜像
这个是我们的zookeeper镜像，没有指定版本默认就是拉取最新的版本

1. docker pull zookeeper

复制代码

kafka镜像 

1. docker pull wurstmeister/kafka

复制代码

由于我们的docker差别容器之间的网络是相互隔开的，以是我们要创建一个共同使用的网络
让差别容器都参加这个网络
docker network create创建我们的网络
然后谁人zookeeper_network是我们自界说的网络名称

1. docker network create --driver bridge zookeeper_network

复制代码

kafka是依靠于zookeeper的以是我们要先安装zookeeper
我们先用run来创建一个zookeeper容器

1. docker run -d --name zookeeper1  --network zookeeper_network -p 2181:2181   zookeeper

复制代码

  -d 是配景运行
  
  --name 是我们自界说容器的名字  我界说的名字是zookeeper1
  
  --network 
  是指定我们的网络环境，我们刚刚创建的网络环境名字叫zookeeper_network，以是我们要让容器参加这个网络
  
  -p 是指定我们的容器袒露给外部的端口  2181：2181是指捏造机（或服务器）的2181端口与容器内部的2181端口做映射
  
  最反面的谁人zookeeper 是我们的使用的镜像源的名称
  一样平常是zookeeper：xxx来实利用用镜像源的版本，如果不指定版本默认用的就是最新版本
  
查察我们创建的网络环境的所在

1. docker inspect zookeeper_network

复制代码

谁人IPv4就是我们的网络环境的所在，这是我的网络环境的所在
我的是12.21.0.2，这个ip所在是要记取方便反面使用的
 
创建一个kafka容器

这段代码有点长，根子自己改吧

1. # 启动kafka
2. docker run -d --name kafka1  --network zookeeper_network -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=<zookeeperIP地址>:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://<宿主机IP地址>:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092  wurstmeister/kafka

复制代码

表明 
   KAFKA_ZOOKEEPER_CONNECT 反面写的是我们的之前的网络的所在
  KAFKA_ADVERTISED_LISTENERS=PLAINTEXT 我们的捏造机（服务器）的本机的所在
  不知道本机所在可以输入 ip addr来查察本机所在

如许子就搭建完成了

---

SpringBoot集成kafka

起首就是springboot和kafka的版本兼容了
Spring for Apache Kafka

然后我们引入两个kafka的依靠 
  

3. <dependency>
4.     <groupId>org.springframework.kafka</groupId>
5.     <artifactId>spring-kafka</artifactId>
6. </dependency>
8. <dependency>
9.     <groupId>org.apache.kafka</groupId>
10.     <artifactId>kafka-clients</artifactId>
11.     <version>3.2.0</version>
12. </dependency>

复制代码

自己对着自己的版原来

自己看b站视频，9分钟就搞定了
63-kafka-集成-Java场景-SpringBoot_哔哩哔哩_bilibili

然后开始写我们的application.yml设置文件

下面是设置文件的全部+分析
着实宁静凡mq差不多
也就是设置生产者和消耗者和一些逾期时间超时时间

重点在于谁人missing-topics-fatal
主题不存在的话，我们是否还要乐成启动
我自己的写的默认的主题是test，但是我还没在kafka内里创建，kafka内里还没有这个叫test的主题
以是我启动的时间，报错然后失败了 
  

1. spring:
2.   kafka:
3.     bootstrap-servers: 192.168.88.130:9092  #Kafka 集群的地址和端口号
4.     producer:
5.       acks: all #生产者发送消息时, Kafka 集群需要确认的确认级别。all 表示需要所有 broker 确认消息已经写入
6.       batch-size: 16384  #生产者在发送消息时, 会先缓存一些消息, 达到 batch-size 后再批量发送。这个参数设置了批量发送的大小。
7.       buffer-memory: 33554432  #生产者用于缓存消息的内存大小
8.       key-serializer: org.apache.kafka.common.serialization.StringSerializer  #定义了消息 key 和 value 的序列化方式。
9.       value-serializer: org.apache.kafka.common.serialization.StringSerializer #定义了消息 key 和 value 的序列化方式。
10.       retries: 0
11.     consumer:
12.       group-id: test #消费者组ID
13.       #消费方式: 在有提交记录的时候，earliest与latest是一样的，从提交记录的下一条开始消费
14.       # earliest：无提交记录,表示从最早的消息开始消费
15.       #latest：无提交记录，从最新的消息的下一条开始消费
16.       auto-offset-reset: earliest  #当消费者没有提交过 offset 时, 从何处开始消费消息
17.       enable-auto-commit: true #是否自动提交偏移量offset
18.       auto-commit-interval: 1s #前提是 enable-auto-commit=true。自动提交 offset 的间隔时间
19.       key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  #定义了消息 key 和 value 的反序列化方式
20.       value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #定义了消息 key 和 value 的反序列化方式
21.       max-poll-records: 2  #一次 poll 操作最多返回的消息数量
22.       properties:
23.         #如果在这个时间内没有收到心跳，该消费者会被踢出组并触发{组再平衡 rebalance}
24.         #消费者与 Kafka 服务端的会话超时时间
25.         session.timeout.ms: 120000
27.         #最大消费时间。此决定了获取消息后提交偏移量的最大时间，超过设定的时间（默认5分钟），服务端也会认为该消费者失效。踢出并再平衡
28.         #消费者调用 poll 方法的最大间隔时间
29.         max.poll.interval.ms: 300000
31.         #消费者发送请求到 Kafka 服务端的超时时间
32.         #配置控制客户端等待请求响应的最长时间。
33.         #如果在超时之前没有收到响应，客户端将在必要时重新发送请求，
34.         #或者如果重试次数用尽，则请求失败。
35.         request.timeout.ms: 60000
37.         #订阅或分配主题时，允许自动创建主题。0.11之前，必须设置false
38.         allow.auto.create.topics: true
40.         #消费者向协调器发送心跳的间隔时间。
41.         #poll方法向协调器发送心跳的频率，为session.timeout.ms的三分之一
42.         heartbeat.interval.ms: 40000
44.         #每个分区里返回的记录最多不超max.partitions.fetch.bytes 指定的字节
45.         #0.10.1版本后 如果 fetch 的第一个非空分区中的第一条消息大于这个限制
46.         #仍然会返回该消息，以确保消费者可以进行
48.         #每个分区最多拉取的消息字节数。
49.         #max.partition.fetch.bytes=1048576  #1M
51.     listener:
52.       #当enable.auto.commit的值设置为false时，该值会生效；为true时不会生效
53.       #manual_immediate:需要手动调用Acknowledgment.acknowledge()后立即提交
54.       #ack-mode: manual_immediate   手动 ACK 的方式。
56.       #如果监听的主题不存在, 是否启动失败。
57.       missing-topics-fatal: false #如果至少有一个topic不存在，true启动失败。false忽略
59.       #消费方式, single 表示单条消费, batch 表示批量消费
60.       #type: single #单条消费？批量消费？ #批量消费需要配合 consumer.max-poll-records
61.       type: batch
63.       #并发消费的线程数
64.       concurrency: 2 #配置多少，就为为每个消费者实例创建多少个线程。多出分区的线程空闲
66.       #默认的主题名称
67.     template:
68.       default-topic: "test"
69.       #springboot启动的端口号
70. server:
71.   port: 9999 #这个是java项目启动的端口

复制代码

根本案例

这是常量类
指定了一个topic和group
主题和分组id
groupid是消耗者组的唯一标识
这个视频9分钟看懂kafka
小朋侪也可以懂的Kafka入门教程，还不快来学_哔哩哔哩_bilibili

生产者

我们这个Autowired主动注入，会根据我们的设置文件的设置来主动注入

1. @Autowired
2.     private KafkaTemplate<String, String> kafkaTemplate;

复制代码

produces内里指定我们前端传的是json格式

 我们往这个标题发送我们的消息，着实这个就是我们的常量类内里写的"test"

消耗者

1. @KafkaListener(topics = SpringBootKafkaConfig.TOPIC_TEST, groupId = SpringBootKafkaConfig.GROUP_ID)
3.     public void topic_test(List<String> messages, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
4.         for (String message : messages) {
6.             //因为这个String是Json，所以我们可以转回Object对象，其实是转成JsonObject对象
7.             final JSONObject entries = JSONUtil.parseObj(message);
8.             System.out.println(SpringBootKafkaConfig.GROUP_ID + " 消费了： Topic:" + topic + ",Message:" + entries.getStr("name"));
9.             //ack.acknowledge();
10.         }
11.     }

复制代码

我们用List<String>来吸收，由于大概一个消耗者吸收多条消息

指定消耗者监听的主题topic
以及指定消耗者的唯一标识GROUP_ID
这些着实都是自己在常量类内里自己写好的


@Header(KafkaHeaders.RECEIVED_TOPIC) String topic
 这个是得到我们的主题topic的名字

我用apifox调试之后，乐成实行了

---

kafka的图形化工具

这里先容一个免费的开源项目KafkaKing
Releases · Bronya0/Kafka-King (github.com)
内里还能指定中文