概述

定义

消息队列

目前企业中比较常见的消息队列产品主要有 Kafka、ActiveMQ、RabbitMQ、
RocketMQ等。

在大数据场景主要采用 Kafka作为消息队列。在 JavaEE开发中主要采用ActiveMQ、RabbitMQ、RocketMQ。

传统消息队列的应用场景

传统的消息队列的主要应用场景包括：缓存/消峰、解耦和异步通信。

• 缓存/消峰：有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况
• 解耦：允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束
  
• 异步通信：允许用户把一个消息放入队列，但并不立即处理它，然后在需要的时候再去处理它们

消息队列的两种模式

Kafka基础架构

（1）Producer：消息生产者，就是向Kafka broker发消息的客户端。
（2）Consumer：消息消费者，向Kafka broker取消息的客户端。
（3）Consumer Group（CG）： 消费者组，由多个consumer组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
（4）Broker：一台 Kafka服务器就是一个broker。一个集群由多个broker组成。一个
broker 可以容纳多个topic。
（5）Topic：可以理解为一个队列，生产者和消费者面向的都是一个topic。
（6）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服
务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。
（7）Replica：副本。一个 topic 的每个分区都有若干个副本，一个 Leader 和若干个
Follower。
（8）Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是Leader。
（9）Follower：每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和Leader 数据的同步。Leader发生故障时，某个Follower会成为新的Leader。

Kafka快速入门

安装部署

集群规划

hadoop102	hadoop103	hadoop104
zk	zk	zk
kafka	kafka	kafka

集群部署

1）官方下载地址：https://kafka.apache.org/downloads.html

2）解压安装包

[yudan@hadoop102 software]$ tar -zxvf kafka_2.12-3.0.0.tgz -C /opt/module/ 

3）修改解压后的文件名称

[yudan@hadoop102 module]$ mv kafka_2.12-3.0.0/ kafka 

4）进入到/opt/module/kafka目录，修改配置文件

[yudan@hadoop102 kafka]$ cd config/ 
[yudan@hadoop102 config]$ vim server.properties

输入以下内容：

#broker的全局唯一编号，不能重复，只能是数字。 
broker.id=0 --需要修改的参数
#处理网络请求的线程数量 
num.network.threads=3 
#用来处理磁盘IO的线程数量 
num.io.threads=8 
#发送套接字的缓冲区大小 
socket.send.buffer.bytes=102400 
#接收套接字的缓冲区大小 
socket.receive.buffer.bytes=102400 
#请求套接字的缓冲区大小 
socket.request.max.bytes=104857600 
#kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔 
log.dirs=/opt/module/kafka/datas --需要修改的参数
#topic在当前broker上的分区个数 
num.partitions=1 
#用来恢复和清理data下数据的线程数量 
num.recovery.threads.per.data.dir=1 
# 每个topic创建时的副本数，默认时1个副本 
offsets.topic.replication.factor=1 
#segment文件保留的最长时间，超时将被删除 
log.retention.hours=168 
#每个segment文件的大小，默认最大1G 
log.segment.bytes=1073741824 
# 检查过期数据的时间，默认5分钟检查一次是否数据过期 
log.retention.check.interval.ms=300000 
#配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理） 
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka --需要修改的参数

5）分发安装包

[yudan@hadoop102 module]$ xsync kafka/

6）分别在hadoop103和hadoop104上修改配置文件/opt/module/kafka/config/server.properties中的broker.id=1、broker.id=2

注：broker.id不得重复，整个集群中唯一。

[yudan@hadoop103 module]$ vim kafka/config/server.properties 
修改: 
# The id of the broker. This must be set to a unique integer for each broker. 
broker.id=1 

[yudan@hadoop104 module]$ vim kafka/config/server.properties 
修改: 
# The id of the broker. This must be set to a unique integer for each broker. 
broker.id=2 

7）配置环境变量

（1）在/etc/profile.d/my_env.sh文件中增加kafka环境变量配置

[yudan@hadoop102 module]$ sudo vim /etc/profile.d/my_env.sh

增加如下内容：

#KAFKA_HOME 
export KAFKA_HOME=/opt/module/kafka 
export PATH=$PATH:$KAFKA_HOME/bin 

（2）刷新一下环境变量

[yudan@hadoop102 module]$ source /etc/profile

（3）分发环境变量文件到其他节点，并source。

[yudan@hadoop102 module]$ sudo /home/atguigu/bin/xsync /etc/profile.d/my_env.sh 
[yudan@hadoop103 module]$ source /etc/profile 
[yudan@hadoop104 module]$ source /etc/profile

8）启动集群

（1）先启动Zookeeper集群，然后启动Kafka。

[yudan@hadoop102   kafka]$ zk.sh start 

（2）依次在hadoop102、hadoop103、hadoop104节点上启动Kafka。

[yudan@hadoop102 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties 
[yudan@hadoop103 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties 
[yudan@hadoop104 kafka]$ bin/kafka-server-start.sh -daemon config/server.properties

注意：配置文件的路径要能够到server.properties。

9）关闭集群

[yudan@hadoop102 kafka]$ bin/kafka-server-stop.sh  
[yudan@hadoop103 kafka]$ bin/kafka-server-stop.sh  
[yudan@hadoop104 kafka]$ bin/kafka-server-stop.sh 

集群启停脚本

1）在/home/atguigu/bin 目录下创建文件kf.sh 脚本文件

[yudan@hadoop102 bin]$ vim kf.sh

脚本如下：

#! /bin/bash 

case $1 in 
"start"){ 
for i in hadoop102 hadoop103 hadoop104 
do 
echo " --------启动 $i Kafka-------" 
ssh $i "/opt/module/kafka/bin/kafka-server-start.sh 
daemon /opt/module/kafka/config/server.properties" 
done 
};; 
"stop"){ 
for i in hadoop102 hadoop103 hadoop104 
do 
echo " --------停止 $i Kafka-------" 
ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh " 
done 
};; 
esac 

2）添加执行权限

[yudan@hadoop102 bin]$ chmod 777 kf.sh

3）启动集群命令

[yudan@hadoop102 ~]$ kf.sh start 

4）停止集群命令

[yudan@hadoop102 ~]$ kf.sh stop

注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止Zookeeper集群。因为Zookeeper集群当中记录着 Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka 集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。

Kafka命令行操作

主题命令行操作

1）查看操作主题命令参数

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh

参数	描述
--bootstrap-server <String: server toconnect to>	连接的 Kafka Broker 主机名称和端口号。
--topic <String: topic>	操作的topic名称。
--create	创建主题。
--delete	删除主题。
--alter	修改主题。
--list	查看所有主题。
--describe	查看主题详细描述。
--partitions <Integer: # of partitions>	设置分区数。
--replication-factor <Integer: replication factor>	设置分区副本。
--config <String: name=value>	更新系统默认的配置。

2）查看当前服务器中的所有topic

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --list

3）创建first topic

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --create --partitions 1 --replication-factor 3 --topic first

• 选项说明：
  • –topic 定义 topic 名
  • –replication-factor 定义副本数
  • –partitions 定义分区数

4）查看first 主题的详情

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first 

5）修改分区数（注意：分区数只能增加，不能减少）

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --alter --topic first --partitions 3 

6）再次查看first主题的详情

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --describe --topic first

7）删除topic

[yudan@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server hadoop102:9092 --delete --topic first

生产者命令行操作

1）查看操作生产者命令参数

[yudan@hadoop102 kafka]$ bin/kafka-console-producer.sh

参数	描述
--bootstrap-server	要连接的 Kafka Broker 主机名称和端口号。
--topic	操作的topic。

2）发送消息

[yudan@hadoop102 kafka]$ bin/kafka-console-producer.sh -bootstrap-server hadoop102:9092 --topic first 

>hello world 
>yudan yudan

消费者命令行操作

1）查看操作消费者命令参数

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh 

参数	描述
--bootstrap-server<String: server toconnect to>	连接的 Kafka Broker 主机名称和端口号。
--topic<String: topic>	操作的topic名称。
--from-beginning	从头开始消费。如果设置此参数，消费者将从该topic的最开始位置开始消费消息
--group <String: consumer group id>	指定消费者组名称

2）消费消息

（1）消费first 主题中的数据

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first

（2）把主题中所有的数据都读取出来（包括历史数据）

[yudan@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first