IT评测·应用市场-qidao123.com

标题: 为什么要用Kafka？单机服务搭建&三步消息交互 [打印本页]
作者: 欢乐狗    时间: 2025-3-13 07:56
标题: 为什么要用Kafka？单机服务搭建&三步消息交互
1、为什么要用Kafka？

1.1、Kafka简介

Apache Kafka是一个分布式流处理平台，最初由LinkedIn开发并于2011年开源。它重要用于解决大规模数据的实时流式处理和数据管道标题。
它的特点包括：高吞吐量、长期化、分布式、实时性。常用于日记收集、消息系统、事件溯源、流处理等场景。
焦点概念：

   官网所在：https://kafka.apache.org/
  1.2、为什么要用Kafka？

以一个典型的日记聚合应用场景为例

该业务场景要求产品的特点：
(1) 数据吞吐量很大： 需要 能够 快速收集各个 渠道的海量日记。
(2) 集群容错性高： 答应集群中少量节点崩溃。
(3) 功能不需要太复杂： Kafka的设计目标是高吞吐、低延迟和可扩展，重要关注消息通报而不是消息处理。以是，并没有支持死信队列、次序消息等高级功能。
(4) 答应少量数据丢失： Kafka本身也在不断优化数据安全标题，目前根本可以以为Kafka可以做到不会丢数据。
2、单机服务搭建

2.1、情况说明&版本选择

运行情况linux ， centos 7.x ，内存2G 4核，JDK 17 (官方要求1.8或以上)
   我的虚拟机目录说明
  所有自定义安装目录：/opt/apps
  服务安装目录：/opt/apps/server
  在server目录下再分别创建两个目录：zookeeper 和 kafka
  Kafka官方下载所在：https://kafka.apache.org/downloads
选择当前最新的3.9.0版本： kafka_2.13-3.9.0.tgz
Zookeeper官方下载所在：https://zookeeper.apache.org/releases.html
Zookeeper的版本并没有逼迫要求，这里我们选择比较新的3.8.4稳定版本。注意下载的是bin版本。
   kafka的安装程序中自带了Zookeeper，Zookeeper的客户端jar包在kafka的安装包的libs目录下。但通常情况下，为了更好维护应用，我们会单独摆设的Zookeeper，而不使用Kafka自带的Zookeeper。
  2.2、安装包上传并解压

把下载好的安装包上传到服务器，并解压。
   上传工具可使用Xftp，客户端毗连工具可使用Xshell，安装包在文末获取，官方个人免费许可，仅需邮箱注册即可。
  如果需要空白centos 7虚拟机，里面已经配置好jdk17和一些常用命令，在文末获取。
  1. # 解压zookeeper
  2. tar -xzvf apache-zookeeper-3.8.4-bin.tar.gz -C /opt/apps/server/zookeeper
  3. # 解压kafka
  4. tar -xzvf kafka_2.13-3.9.0.tgz -C /opt/apps/server/kafka/
复制代码
2.3、配置情况变量

将摆设目录下的bin目录路径配置到path情况变量中。
  1. # 打开系统级配置（对所有用户生效）
  2. vim /etc/profile
  4. # 配置的内容如下
  5. export ZOOKEEPER_HOME=/opt/apps/server/zookeeper/apache-zookeeper-3.8.4-bin
  6. export PATH=$ZOOKEEPER_HOME/bin:$PATH
  7. export KAFKA_HOME=/opt/apps/server/kafka/kafka_2.13-3.9.0
  8. export PATH=$KAFKA_HOME/bin:$PATH
  10. # 添加内容后，保存退出 【:wq!】
  11. # 生效环境变量
  12. source /etc/profile
  13. # 查看环境变量
  14. echo $PATH
复制代码
2.4、单机服务搭建

启动Kafka之前需要先启动Zookeeper。启动Zookeeper前需要先做一下简单配置。
进入Zookeeper摆设目录，切换到conf目录，复制zoo_sample.cfg，修改配置dataDir值，默认是暂时目录，随时会被删撤除。
  1. # 切换到Zookeeper的配置目录
  2. cd /opt/apps/server/zookeeper/apache-zookeeper-3.8.4-bin/conf/
  3. # 复制出一个配置文件
  4. cp zoo_sample.cfg zoo.cfg
  5. # 修改配置文件
  6. vim zoo.cfg
  7. # 修改内容如下
  8. dataDir=/opt/apps/server/zookeeper/data
复制代码
启动Zookeeper
  1. # 切换到Zookeeper根目录
  2. cd $ZOOKEEPER_HOME
  3. # 启动命令
  4. nohup bin/zkServer.sh start conf/zoo.cfg &
复制代码
确认服务是否启动成功，有三种方式
  1. # 方式一：jps指令，成功可看到一个QuorumPeerMain进程
  2. # 方式二：连接客户端测试，Zookeeper默认端口是2181
  3. bin/zkCli.sh -server localhost:2181
  4. # 方式三：检查端口是否监听
  5. netstat -tuln | grep 2181
复制代码
启动Kafka
  1. # 切换到Kafka根目录
  2. cd $KAFKA_HOME
  3. # 启动命令
  4. nohup bin/kafka-server-start.sh config/server.properties &
复制代码
确认服务是否启动成功，有两种方式
  1. # 方式一：jps指令，成功可看到一个Kafka进程
  2. # 方式二：检查端口是否监听，Kafka默认端口是9092
  3. netstat -tuln | grep 9092
复制代码
3、三步消息交互

Kafka的底子工作机制是消息发送者将消息发送到Kafka上指定的topic，消息消耗者从指定的topic上消耗消息。

底子消息交互三步骤：
第一步：使用Kafka提供的客户端脚本创建Topic。
第二步：启动一个消息发送者端，向指定名称的Topic发送消息。
第三步：启动一个消息消耗端，从指定名称的Topic上吸收消息。
此中，生产和消耗者并不需要同时启动。它们之间可以进行数据交互，但并不依赖于对方。没有生产者，消耗者依然可以正常工作。反过来，没有消耗者，生产者也可以正常工作。表现了生产者和消耗者之间的解耦。
3.1、实验消息交互

第一步：创建Topic名为test
  1. # 创建Topic
  2. bin/kafka-topics.sh --create --topic test --bootstrap-server localhost:9092
  3. # 创建成功会提示信息
  4. Created topic test
  5. # 查看topic
  6. bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092
复制代码
创建成功时，列出数据示例

第二步：启动一个消息发送者端，向名为test的Topic发送消息。
  1. bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
复制代码
当命令行出现【 > 】符号后，恣意输入一些字符，按回车键完成消息发送操作。Ctrl+C 退出命令行。
第三步：启动一个消息消耗端，从名为tes的Topic上吸收消息。
  1. # 不加任何参数，是指默认从最新消息（latest） 开始消费
  2. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
  3. test1
  4. test2
  5. 123
  6. ^CProcessed a total of 3 messages
复制代码
这样就完成了一个底子的消息交互。
3.2、其他消耗模式

消息消耗时，如果不指定参数，默认从最新消息开始消耗。消息消耗还有另外两种模式：指定消耗进度和分组消耗。
指定消耗进度
又可以分为重新开始或精确到从哪一条消息之后开始。
  1. # 如果想要消费之前发送的消息，可以加参数--from-beginning指定
  2. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
  4. # 如果需要更精确的消费消息，甚至可以指定从哪一条消息开始消费。
  5. # 从第0个partition上的第4个消息开始读
  6. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset 4
复制代码
分组消耗
对于每个消耗者，可以指定一个消耗者组。
kafka中的同一条消息，只能被同一个消耗者组下的某一个消耗者消耗。不属于同一个消耗者组的其他消耗者，可以消耗到这一条消息。
在kafka-console-consumer.sh脚本中，可以通过参数–consumer-property group.id=【消耗组名】指定所属的消耗者组。
实验时，可以启动两个消耗者组、三个消耗者实例【三个窗口】，验证 分组消耗机制。
  1. # 在窗口1和窗口2，输入消费者组【testGroup】的消费命令
  2. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer-property group.id=testGroup
  3. # 在窗口3，输入消费者组【testGroup2】的消费命令
  4. bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer-property group.id=testGroup2
复制代码
一切预备停当后，在生产者端发送消息，发现【窗口1】没有消耗提示。由此，可证实上面的结论。

查看消耗者组的偏移量
可以使用kafka-consumer-groups.sh观测消耗者的情况，包括它们的消耗进度。
  1. bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group testGroup
复制代码

从这里可以看出，固然业务上是通过Topic分发消息的，但是实际上，消息是保存在Partition数据结构上的。
4、我的公众号&资料获取

敬请关注我的公众号：大象只为你，连续更新技能知识…
相关资料获取：
如需centos7空白虚拟机，请后台复兴：blankOS。
blankOS登录账号暗码：root / 123456
如需客户端毗连工具Xshell，请后台复兴：Xshell。
如需上传工具Xftp，请后台复兴：Xftp。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。



欢迎光临 IT评测·应用市场-qidao123.com (https://dis.qidao123.com/) Powered by Discuz! X3.4