目录
一、docker安装和配置Kafka
1.拉取 Zookeeper 的 Docker 镜像
2.运行 Zookeeper 容器
3.拉取 Kafka 的 Docker 镜像
4.运行 Kafka 容器
5.下载 Kafdrop
6.运行 Kafdrop
7.如果docker pull wurstmeister/zookeeper或docker pull wurstmeister/kafka下载很慢，可以找一台网络比较好的机器，输入这两个命令进行下载，下载后使用docker save -o保存为tar文件，然后将tar文件传输到目标机器后，使用docker load -i加载tar文件为docker镜像文件
8.使用 Kafka 自带的工具来创建一个名为 users 的主题
9.验证 Kafka,可以使用 Kafka 自带的工具来验证 Kafka 是否正常工作。例如，启动一个 Kafka 消费者来监听 users 主题：
二、在Spring Boot项目中集成和使用Kafka
1. 添加依赖
2. 配置Kafka
3. 创建消息对象
4. 创建生产者
5. 创建消费者
6. 测试
三、web访问Kafdrop
一、docker安装和配置Kafka
1.拉取 Zookeeper 的 Docker 镜像
docker pull wurstmeister/zookeeper
2.运行 Zookeeper 容器
docker run -d --name zookeeper -p 2181:2181 wurstmeister/zookeeper
3.拉取 Kafka 的 Docker 镜像
docker pull wurstmeister/kafka
4.运行 Kafka 容器
docker run -d --name kafka -p 9092:9092 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.7.46:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 --link zookeeper:zookeeper wurstmeister/kafka
5.下载 Kafdrop
docker pull obsidiandynamics/kafdrop
6.运行 Kafdrop
docker run -p 9000:9000 -d --name kafdrop -e KAFKA_BROKERCONNECT=192.168.7.46:9092 -e JVM_OPTS="-Xms16M -Xmx48M" obsidiandynamics/kafdrop
7.如果docker pull wurstmeister/zookeeper或docker pull wurstmeister/kafka下载很慢，可以找一台网络比较好的机器，输入这两个命令进行下载，下载后使用docker save -o保存为tar文件，然后将tar文件传输到目标机器后，使用docker load -i加载tar文件为docker镜像文件
下载：
docker pull wurstmeister/zookeeper
docker pull wurstmeister/kafka
(kafdrop是一个kafka的web图形管理界面)
docker pull obsidiandynamics/kafdrop
打包：
docker save -o ./zookeeper.tar wurstmeister/zookeeper
docker save -o ./kafka.tar wurstmeister/kafka
docker save -o ./kafdrop.tar obsidiandynamics/kafdrop
传输：
scp kafka.tar root@192.168.7.46:/usr/root/kafka 回车后输入密码即可
scp zookeeper.tar root@192.168.7.46:/usr/root/kafka 回车后输入密码即可
scp kafdrop.tar root@192.168.7.46:/usr/root/kafka 回车后输入密码即可
目标机加载成docker镜像
docker load -i /usr/root/kafka/kafka.tar
docker load -i /usr/root/kafka/zookeeper.tar
docker load -i /usr/root/kafka/kafdrop.tar
查看镜像列表
docker images
8.使用 Kafka 自带的工具来创建一个名为 users 的主题
docker exec -it kafka kafka-topics.sh --create --topic users --partitions 1 --replication-factor 1 --bootstrap-server 192.168.7.46:9092
9.验证 Kafka,可以使用 Kafka 自带的工具来验证 Kafka 是否正常工作。例如，启动一个 Kafka 消费者来监听 users 主题：
docker exec -it kafka kafka-console-consumer.sh --topic users --from-beginning --bootstrap-server 192.168.7.46:9092
这个命令，会启动一个额外的 Kafka 消费者来监听 users 主题。这个消费者是通过 Kafka 自带的 kafka-console-consumer.sh 工具启动的，主要用于测试和验证目的。它会持续监听并打印出发送到 users 主题的所有消息。
二、在Spring Boot项目中集成和使用Kafka
1. 添加依赖
首先，在你的pom.xml文件中添加Kafka的依赖：
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
2. 配置Kafka
在application.properties或application.yml文件中配置Kafka的相关属性。这里以application.properties为例：
# Kafka broker地址
spring.kafka.bootstrap-servers=localhost:9092
# 生产者配置
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
# 消费者配置
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*
3. 创建消息对象
假设我们要发送和接收一个简单的KafkaMsgs 对象：
public class KafkaMsgs {
private String id;
private String msg;
private Long date;
// 构造函数、getter和setter省略
}
4. 创建生产者
创建一个生产者类来发送消息：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducer {
@Autowired
private KafkaTemplate<String, KafkaMsgs> kafkaTemplate;
public void sendMessage(String topic, KafkaMsgs kafkaMsgs) {
kafkaTemplate.send(topic, kafkaMsgs);
}
}
5. 创建消费者
创建一个消费者类来接收消息：
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumer {
@KafkaListener(topics = "users", groupId = "my-group")
public void listen(KafkaMsgs kafkaMsgs) {
System.out.println("Received message: " + kafkaMsgs);
}
}
6. 测试
你可以创建一个简单的测试类来验证生产和消费是否正常工作：
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.esop.resurge.core.config.kafka.KafkaProducer;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.airbubble.kingdom.army.reponse.FeedBack;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;import java.util.Date;
@Api(tags="kafka数据控制器")
@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {
@Autowired
KafkaProducer kafkaProducer;@ApiOperation(value = "测试发送数据到kafka", httpMethod = "GET")
@GetMapping(value = "/sendKafkaData")
public FeedBack<String> sendKafkaData(
@ApiParam(value = "topic", required = true) @RequestParam(required = true,value = "topic") String topic,
@ApiParam(value = "msg", required = true) @RequestParam(required = true,value = "msg") String msg
) throws Exception {
kafkaProducer.sendMessage(topic, new com.esop.resurge.core.config.kafka.KafkaMsgs(
IdUtil.fastUUID(),
msg,
Long.valueOf(DateUtil.format(new Date(), DatePattern.NORM_DATETIME_FORMAT).replace(" ", "").replace(":", "").replace("-", ""))
));
return FeedBack.getInstance("发送成功");
}}
三、web访问Kafdrop
打开浏览器，访问 http://192.168.7.46:9000，你应该能够看到 Kafdrop 的 Web 界面
