Linux环境下Kafka的安装与使用(SpringBoot整合云服务器上的Kafka)

本文我们主要讲 Kafka 的安装和使用，对于 Kafka 的简介和功能作用就不再详细介绍，给大家推荐一篇文章，可以自行了解：https://blog.csdn.net/weixin_42238150/article/details/107449787

1. 下载kafka

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

下载地址：http://archive.apache.org/dist/kafka/2.3.1/

因为国外网址，下载速度很慢，建议大家使用镜像地址下载：https://www.apache.org/dyn/closer.cgi?path=/kafka/2.5.1/kafka_2.13-2.5.1.tgz

镜像下载链接：https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/2.5.1/kafka_2.13-2.5.1.tgz

2. 在Linux下安装Kafka

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

我这里放到：/usr/local/src/software下

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

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

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

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

进入bin目录查看：

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

zookeeper.properties

首先我们在/usr/local/src/software/kafka_2.13-2.5.1 下新建一个 data 文件夹，用于存放zookeeper运行后的数据：mkdir data，创建成功后进入该文件夹并执行pwd，然后拷贝路径地址/usr/local/src/software/kafka_2.13-2.5.1/data

然后进入 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 files
log.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 时候，消费者扔继续会读取消息！

参考文章：https://blog.csdn.net/qq_29308413/article/details/84893539

如果启动 Kafka 报错：The broker is trying to join the wrong cluster. Configured zookeeper.connect may be wrong.，参考文章：https://www.cnblogs.com/qi-yuan-008/p/13912143.html

5. SpringBoot 整合Kafka

新建一个Spring Boot 项目：

5.1 引入依赖

pom.xml 中引入Kafka依赖：

<!-- https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.3.7.RELEASE</version>
</dependency>

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 测试使用

生产者

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

消费者

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

Test测试

@SpringBootTest
public 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://blog.csdn.net/chang_harry/article/details/103501369