开发者社区> 兴趣使然的草帽路飞> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

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

简介: 0. 前言
+关注继续查看

本文我们主要讲 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/

image.png

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

image.png

镜像下载链接: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 传输到云服务器/虚拟机指定位置

image.png

我这里放到:/usr/local/src/software

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

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

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

image.png

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

进入bin目录查看:

image.png

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


image.png

修改完成后,wq 保存退出即可!

server.properties

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


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


image.png

image.png

我们将其修改为:(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 的启动日志文件


image.png

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

image.png

上图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 命令,创建一个主题


image.png

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

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


image.png

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

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

image.png

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

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

image.png

我们再从生产者生产一条消息 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();
        }
    }
}

测试结果

image.png

踩坑:连远程Kafka报错UnknownHostException

这个问题参考这篇文章即可解决:https://blog.csdn.net/chang_harry/article/details/103501369


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
vos3000怎样解决费率表导入时精度丢失,Linux 硬件信息查询命令是怎样的
vos3000怎样解决费率表导入时精度丢失,Linux 硬件信息查询命令是怎样的
11 0
【Linux】之【CPU】相关的命令及解析[lscpu、mpstat]
lscpu命令从sysfs和/proc/cpuinfo收集cpu体系结构信息,命令的输出比较易读,命令输出的信息包含cpu数量,线程,核数,套接字,缓存等。
24 0
【Linux】之【内存】相关的命令&&解析以及内存相关的问题[free、meminfo、内存泄漏、内存溢出、Overcommit]
free命令的功能是显示系统内存使用量情况,包含物理和交换内存的总量、使用量和空闲量情况。
16 0
【Linux】之【磁盘】相关的命令及解析[df、du、iostat、iotop]
df命令来自于英文词组”Disk Free“的缩写,其功能是用于显示系统上磁盘空间的使用量情况。df命令显示的磁盘使用量情况含可用、已有及使用率等信息,默认单位为Kb,建议使用-h参数进行单位换算。
12 0
【Linux】之【网络】相关的命令及解析[ethtool、nload、nethogs、iftop、iptraf、ifstat]
ethtool命令用于查询ethX网口基本设置、及设置网卡的参数。
25 0
【Linux】综合性命令及解析【top、htop、vmstat、dstat、glances、sar】
top命令的功能是用于实时显示系统运行状态,包含处理器、内存、服务、进程等重要资源信息。运维工程师们常常会把top命令比作是“加强版的Windows任务管理器”,因为除了能看到常规的服务进程信息以外,还能够对处理器和内存的负载情况一目了然,实时感知系统全局的运行状态,非常适合作为接手服务器后执行的第一条命令。
36 0
Linux—history命令 –– 查看与操纵历史命令
history命令用于显示用户以前执行过的历史命令,并且能对历史命令进行追加和删除等操作。
10 0
+关注
兴趣使然的草帽路飞
我是路飞,一个双非本科出身的Java程序员,22届校招收获阿里、美团、哔哩哔哩意向书,目前是阿里大家庭中的一名全栈开发工程师。我的博客地址(CSDN 10万粉丝博主、博客专家):csp1999.blog.csdn.net
87
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载
冬季实战营第三期:MySQL数据库进阶实战
立即下载