本文共 3805 字，大约阅读时间需要 12 分钟。

0. 前言

本文我们主要讲 Kafka 的安装和使用，对于 Kafka 的简介和功能作用就不再详细介绍，给大家推荐一篇文章，可以自行了解：

1. 下载kafka

• Kafka 官网：
• 下载地址：

因为国外网址，下载速度很慢，建议大家使用镜像地址下载：

镜像下载链接：

2. 在Linux下安装Kafka

2.1 使用Xftp 将Kafka 传输到云服务器/虚拟机指定位置

我这里放到：下

2.2 使用Xshell 连接云服务器/虚拟机

进入放置 Kafka 压缩包的文件夹下，解压压缩包：tar -zxvf kafka_2.13-2.5.1.tgz

然后cd 命令进入解压后的文件夹：

• bin 目录中存放一些可执行的命令，比如启动关闭等
• config 目录中存放相关配置文件
• libs 目录存放一些依赖包

进入bin目录查看：

2.3 在config目录中修改相关配置

zookeeper.properties

• 首先我们在 下新建一个 data 文件夹，用于存放zookeeper运行后的数据：mkdir data，创建成功后进入该文件夹并执行pwd，然后拷贝路径地址
• 然后进入 config 目录修改 zookeeper.properties 配置文件：vim zookeeper.properties

• 修改完成后，wq 保存退出即可！

server.properties

接下来我们修改下 server.properties 配置文件，同样在 config 目录下修改配置文件 vim server.properties

在 vim 中收入/log.dir 快速定位到要修改的位置：

我们将其修改为：(data 路径还是刚才复制的路径)

# A comma separated list of directories under which to store log fileslog.dirs=/usr/local/src/software/kafka_2.13-2.5.1/data/kafka-logs

然后wq 保存退出即可！

3. 启动Kafka

3.1 先启动 Zookeeper 再启动Kafka

因为 Kafka 依赖于 Zookeeper 所以先启动后者！

• 进入 bin 目录下执行 ：./zookeeper-server-start.sh ../config/zookeeper.properties 命令，以指定配置文件启动 zookeeper 服务，守护进程启动方式，在结尾加一个 & 即可；
  • zookeeper 服务启动后，打开一个新的命令窗口，执行./kafka-server-start.sh ../config/server.properties 命令，启动 Kafka服务，守护进程启动方式，在结尾加一个 & 即可；
  • 然后我们可以在刚才新建的 data 目录下看到 Kafka 和 zookeeper 的启动日志文件

注意：如果使用的是云服务器，提前开放安全组端口：

上图2189 和 9092 是我的云服务器开放的端口！

如果使用的是虚拟机请不要忘记防火墙放行端口！

3.2 关闭zookeeper 和 kafka

• 在 bin 目录下关闭 zookeeper ./zookeeper-server-stop.sh

• 关闭kafka ./kafka-server-stop.sh

4. 使用Kafka

这里我们以Kafka消息发布订阅(topics)功能为例，来启动相关的文件

4.1 创建主题

进入 bin 目录，执行 ./kafka-topics.sh --create --bootstrap-server xxx.xxxx.xxx:9092 --replication-factor 1 --partitions 1 --topic test 命令，创建一个主题

4.2 查看所有创建的主题列表

执行命令：./kafka-topics.sh --list --bootstrap-server xxx.xxxx.xxx:9092

4.3 生产者(producer)向Kafka某个主题上发送消息

执行命令：./kafka-console-producer.sh --broker-list xxx.xxxx.xxx:9092 --topic test

4.4 消费者(consumer)从Kafka某个主题上接收消息

执行命令：./kafka-console-consumer.sh --bootstrap-server 8.131.66.136:9092 --topic test --from-beginning

当我们再从生产者生产一条消息 haha 时候，消费者扔继续会读取消息！

参考文章：

如果启动 Kafka 报错：The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.，参考文章：

5. SpringBoot 整合Kafka

新建一个Spring Boot 项目：

5.1 引入依赖

pom.xml 中引入Kafka依赖：

   
       
    
     org.springframework.kafka
        
    
     spring-kafka
        
    
     2.3.7.RELEASE
    
   

5.2 配置Kafka

# spring 相关配置spring:  # kafka 配置  kafka:    # 服务器地址及其 kafka 端口    bootstrap-servers: 8.131.66.136:9092    # 消费者配置    consumer:      # 消费者分组的id,可以在 kafka 的 config 目录下的 consumer.properties 中配置      group-id: test-consumer-group      # 是否自动提交消费者的偏移量      enable-auto-commit: true      # 自动提交的频率 3s      auto-commit-interval: 3000

5.3 测试使用

生产者

/** * 生产者 */@Componentclass KafkaProducer {
       @Autowired    private KafkaTemplate kafkaTemplate;// 注入kafkaTemplate    /**     * 发送消息     *     * @param topic   消息主题     * @param content 消息内容     */    public void sendMessage(String topic, String content) {
           kafkaTemplate.send(topic, content);    }}

消费者

/** * 消费者 */@Componentclass KafkaConsumer {
       /**     * 消费者订阅的主题为test     * 就是通过 @KafkaListener(topics = {"test"}) 注解实现的     *     * @param record 接收的消息被封装成 ConsumerRecord 对象     */    @KafkaListener(topics = {
   "test"})    public void handleMessage(ConsumerRecord record) {
           System.out.println(record.value());    }}

Test测试

@SpringBootTestpublic class KafkaTests {
       @Autowired    private KafkaProducer kafkaProducer;// 注入生产者kafkaProducer    @Test    public void testKafka() {
           // 生产者发送消息        kafkaProducer.sendMessage("test", "Hello 你好!");        kafkaProducer.sendMessage("test", "在吗？");        // 在这里进行一下线程阻塞，模仿消费者消费消息的过程        try {
               Thread.sleep(1000 * 3);// 10s        } catch (InterruptedException e) {
               e.printStackTrace();        }    }}

测试结果：

踩坑：连远程Kafka报错UnknownHostException

这个问题参考这篇文章即可解决：

转载地址：https://csp1999.blog.csdn.net/article/details/110310012 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！