【星环云课堂大数据实验】kafka消息发布与订阅

文章目录

- 一、Kafka概述
- 二、实验环境
- 三、实验准备
- 四、实验目的
- 五、实验步骤
- - 5.1、创建Kafka Topic
  - 5.2、Kafka消息发布
  - 5.3、Kafka消息订阅
- 六、实验感悟

一、Kafka概述

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。该项目的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。其持久化层本质上是一个“按照分布式事务日志架构的大规模发布/订阅消息队列”，[4]这使它作为企业级基础设施来处理流式数据非常有价值。此外，Kafka可以通过Kafka Connect连接到外部系统（用于数据输入/输出），并提供了Kafka Streams——一个Java流式处理库。
大白话就是kafka其实消息队列得一种。

二、实验环境

基于星环云课堂TranswarpVD

三、实验准备

查看Zookeeper和Kafka集群的服务地址
打开Transwarp Manager查看Zookeeper集群和kafka的服务地址如下所示
zookeeper地址：
172.18.48.5:2181
172.18.48.6:2181
172.18.48.7:2181
kafka地址**（也就是Broker地址）**：
172.18.48.5:9092
172.18.48.6:9092
172.18.48.7:9092
172.18.48.8:9092

后续的命令替换
{Broker_List} ：172.18.48.5:9092,172.18.48.6:9092,172.18.48.7:9092，172.18.48.8:9092
{Zookeeper_List}： 172.18.48.5:2181,172.18.48.6:2181，172.18.48.7:2181
{topic _name} ：kafka_topic_zhanghaodong

在这里插入图片描述

四、实验目的

• 掌握Kafka的基本使用。‘
• 了解Kafka消息的发布\订阅机制。

五、实验步骤

5.1、创建Kafka Topic

• 任务：利用TDH Client集成的Kafka客户端，创建名为{topic_name}的Kafka Topic，命名规范为kafka_topic_账号，如账号student1，则创建Topic为kafka_topic_student1。
• 步骤
Linux：

cd /transwarp/Desktop/TDH-Client/kafka/bin
// 进入TDH Client集成的Kafka客户端目录
./kafka-topics.sh --bootstrap-server {Broker_List} --create --topic {topic_ name} --partitions 1 --replication-factor 1 --command-config /transwarp/Desktop/TDH-Client/kafka/config/consumer.properties
// 创建Topic
./kafka-topics.sh --zookeeper {Zookeeper_List} --list
// 列出所有Topic，查看Topic是否创建成功

5.2、Kafka消息发布

• 任务：在当前命令行窗口中，创建Kafka Console Producer，它负责在控制台向Kafka Topic发布消息。
• 步骤
Linux：

./kafka-console-producer.sh --topic {topic _name} --broker-list {Broker_List} --producer.config /transwarp/Desktop/TDH-Client/kafka/config/producer.properties
// 创建Console Console Producer

5.3、Kafka消息订阅

• 任务：新建一个命令行窗口，创建Kafka Console Consumer，它负责在控制台接收Kafka Topic的消息。
• 说明：在Producer窗口中输入消息并回车，消息发送到Kafka Topic，Consumer窗口可实时接收到消息。
• 步骤
Linux：

执行TDH Client的init.sh脚本，启动TDH Client
source /transwarp/Desktop/TDH-Client/init.sh
创建Kafka Console Consumer
cd /transwarp/Desktop/TDH-Client/kafka/bin

./kafka-console-consumer.sh --topic {topic_ name} --from-beginning --bootstrap-server {Broker_List} --consumer.config /transwarp/Desktop/TDH-Client/kafka/config/consumer.properties
在控制台生产者窗口中输入以下数据
Hello World
Hello Kafka
返回控制台消费者窗口，可以看到消息已接收到