1. 消息队列基本介绍
1. 什么是消息队列?
消息Message :比如两个设备进行数据的传输,所传输的数据,都可以称为消息,可以是文
本、音频
队列Queue :是一种”先进先出 的数据结构。类似排队买票、羽毛球筒。
消息队列MQ :是用来保存消息的一个容器。 消息队列需要两个功能接[ 7供外部调用,一个
是生产一个是消费。主要是进行数据存储和读取。把数据放到消息队列叫做生产者,从队列里
取数据叫做消费者。
2. MQ.主要分为两类点对点发布/订阅
共同点:
消息的生产者(Producer)生产消息发送到队列中,然后消息的消费者(Consumer)从队列中读取并消费消息。
不同点:
点对点:消息队列(Queue)、发送者(Sender)、接收者(Receiver)
一个生产者生产的消息只能有一个消费者,消息一旦被消费,消息就不在消息队列中了,比如:
钉钉的澡堂模式、打电话等。都是消息发送到消息队列后只能被一个接收者接收, 当接收完毕消息则销毁。
发布/订阅:消息队列(Queue)、发布者(Publisher)、 订阅者(Subscriber)、 主题( Topic)
每个消息可以有多个消费者,彼此互不影响,比如:我使用公众号发布一篇文章,关注我的人都
能看到,即发布到消息队列的消息能被多个接收者(订阅者)接收。
2. 消息队列使用场景
- 1.解耦
- 2.异步
- 3.削峰
3. kafka介绍
3.1 什么是kafka
kafka是一个实时数据处理系统。 实时数据处理系统就是数据一旦产生 ,I就要能快速进行处理的系统。对于实时数据处理系统,最常见的就是消息队列。kafka 也是一1MQ消息队列。
3.2 kafka的特点
高吞吐量:可以满足每秒百万级别消息的生产和消费。
持久性:有一套完善的消息存储机制,确保数据的高效安全的持久化。
分布式:基于分布式的扩展和容错机制,当某一台发生故障失效时 ,可以实现故障自动转移。
3.3 Kafka架构
Broker : kafka集群中包含多个kafka服务掉,每一个kafka服务节点就称为一个broker
Topic:主题, Kafka将消息分J别类，每-类的消息称之为( Topic).(Kafka消息数据是存储在硬盘上的
Partition:分区,每个Topic包含一个或多 个Partition ,在创建Topic时指定包含的Partition数量(目的是为了进行分布式存储
Replication:副本,每个分区可以有多副本,分布在不同的Broker上,选出一个副本作为Leader ,所有的读写
请求都会通过Leader完成,但Follower只负责备份数据,所有Follower会自动的从Leader中复制数据,当Leader
宕机后,会从Follower中选出一个 新的Leader继续提供服务,实现故障自动转移。
Message:消息,是通信的基本单位,每个消息都属于-个Partition
Producer:消息的生产者,向Kafka的一个topic发布消息
Consumer:消息的消费者,订阅topic并读取其发布的消息
Consumer Group:每个Consumer属于-个特定的Consumer Group ,多个Consumer可以属于同一个ConsumerGroup中
Zookeeper:,要用来存储Kafka的元数据信息 ,比如:有多少集群节点主题名称、主要用来协调kafka的正常运行。但发送给Topic本身的消息数据并不存储在ZK中,而存在kafka的磁盘文件中。
4. kafka基本使用
1. 安装zookeeper
[root@kafka--node1 ~]# yum install java maven -y
[root@kafka-node1 ~]# tar xf apache-zookeeper-3.5.6-bin.tar.gz
[root@kafka-node1 ~]# cd apache-zookeeper-3.5.6-bin/conf/
[root@kafka-node1 conf]# cp zoo_sample.cfg zoo.cfg
[root@kafka-node1 conf]# grep -v "^#" zoo.cfg
tickTime=2000
initLimit=10
syncLimit=5
dataDir=../data
clientPort=2181[root@kafka-node1 conf]# mkdir ../data
[root@kafka-node1 conf]# cd ../bin/
[root@kafka-node1 bin]# ./zkServer.sh start
[root@kafka-node1 bin]# ./zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.5.6-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower2. 连接测试：
[root@kafka-node1 bin]# ./zkCli.sh
WatchedEvent state:SyncConnected type:None path:null[zk: localhost:2181(CONNECTED) 0]
[zk: localhost:2181(CONNECTED) 0]
[zk: localhost:2181(CONNECTED) 0] 3. 安装kafka-
[root@kafka-node1 ~]# tar xf kafka-_2.12-2.3.0.tgz
[root@kafka-node1 ~]# cd /opt/kafka-_2.12-2.3.0/bin/[root@kafka-node1 bin]# ./kafka--server-start.sh ../config/server.properties
检测9092（kafka-）端口是否开启
检测2181（zookeeper）端口是否开启[root@kafka-node1 ~]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /
[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, latest_producer_id_block, log_dir_event_notification, zookeeper]
[zk: localhost:2181(CONNECTED) 2] ls /brokers
[ids, seqid, topics][root@kafka-node1 bin]# jps
6436 kafka-
1066 cerebro.cerebro-0.8.5-launcher.jar
11739 Jps
3071 QuorumPeerMain4. 使用kafka-创建topics
[root@kafka-node1 bin]# cd /opt/kafka-_2.12-2.3.0/bin/
[root@kafka-node1 bin]# ./kafka--topics.sh \
> --create \
> --zookeeper localhost:2181 \
> --replication-factor 1 \
> --partitions 3 \
> --topic cheng
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
Created topic cheng.5. 查看topic详情
[root@kafka-node1 bin]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkCli.sh
[zk: localhost:2181(CONNECTED) 1] ls /brokers/topics/cheng/partitions
[0, 1, 2]6. producer模拟生产者，产生数据
[root@kafka-node1 bin]# ./kafka--console-producer.sh --broker-list localhost:9092 --topic cheng7. consumer模拟消费者，模拟消费
[root@kafka-node1 bin]# ./kafka--console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic cheng \
--from-beginning8. 删除topic
[root@kafka-node1 bin]# ./kafka--topic.sh \
--delete \
zookeeper localhost:2181 \
--topic cheng \
5. kafka集群部署
1.安装zookeeper集群
10.0.0.161
10.0.0.162
10.0.0.163
https://www.jianshu.com/p/1a99eebdf0751.安装java
yum install -y java maven2.配置zookeeper
[root@kafka-node1 conf]# cat zoo.cfg
# 服务器之间或客户端与服务器之间维持心跳的时间间隔
# tickTime以毫秒为单位。
tickTime=2000# 集群中的follower服务器(F)与leader服务器(L)之间的初始连接心跳数
initLimit=10# 集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数
syncLimit=5# 数据保存目录
dataDir=../data
# 日志保存目录
dataLogDir=../logs# 客户端连接端口
clientPort=2181# 客户端最大连接数。# 根据自己实际情况设置，默认为60个
maxClientCnxns=60# 三个接点配置，格式为： server.服务编号=服务地址、LF通信端口、选举端口
server.1=10.0.0.161:2888:3888
server.2=10.0.0.162:2888:3888
server.3=10.0.0.163:2888:38883.创建数据存储目录
[root@kafka-node1 conf]# mkdir ../data4.将配置好的zookeeper拷贝至另外两个节点
[root@kafka-node1 opt]# scp -rp apache-zookeeper-3.5.6-bin root@10.0.0.162:/opt
[root@kafka-node1 opt]# scp -rp apache-zookeeper-3.5.6-bin root@10.0.0.163:/opt5.在三个节点上，分别在/opt/apache-zookeeper-3.5.6-bin/data/myid写入节点标记：
node1的操作
[root@kafka-node1 opt]# echo "1" > /opt/apache-zookeeper-3.5.6-bin/data/myidnode2的操作
[root@kafka-node2 ~]# echo "2" > /opt/apache-zookeeper-3.5.6-bin/data/myidnode3的操作
[root@kafka-node3 ~]# echo "3" > /opt/apache-zookeeper-3.5.6-bin/data/myid6.启动zookeeper集群
[root@kafka-node1 ~]# cd /opt/apache-zookeeper-3.5.6-bin/bin/
[root@kafka-node1 bin]# ./zkServer.sh start7.检查集群状态
[root@kafka-node3 bin]# ./zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/apache-zookeeper-3.5.6-bin/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader8. kafka集群安装0.初始化环境
[root@kafka-node1 logs]# cat /etc/hosts
10.0.0.161 kafka-node1
10.0.0.162 kafka-node2
10.0.0.163 kafka-node3 1.解压kafka压缩包
[root@kafka-node1 ~]# cp kafka_2.12-2.3.0.tgz /opt/
[root@kafka-node1 ~]# cd /opt/
[root@kafka-node1 opt]# tar xf kafka_2.12-2.3.0.tgz 2.配置kafka
[root@kafka-node1 config]# cat server.properties
############################# Server Basics #############################
# broker的id，值为整数，且必须唯一，在一个集群中不能重复
broker.id=1############################# Socket Server Settings #############################
# kafka默认监听的端口为9092 (默认与主机名进行连接)
listeners=PLAINTEXT://10.0.0.161:9092# 处理网络请求的线程数量，默认为3个
num.network.threads=3# 执行磁盘IO操作的线程数量，默认为8个
num.io.threads=8# socket服务发送数据的缓冲区大小，默认100KB
socket.send.buffer.bytes=102400# socket服务接受数据的缓冲区大小，默认100KB
socket.receive.buffer.bytes=102400# socket服务所能接受的一个请求的最大大小，默认为100M
socket.request.max.bytes=104857600############################# Log Basics #############################
# kafka存储消息数据的目录
log.dirs=../data# 每个topic默认的partition数量
num.partitions=3# 在启动时恢复数据和关闭时刷新数据时每个数据目录的线程数量
num.recovery.threads.per.data.dir=1############################# Log Flush Policy ############################# # 消息刷新到磁盘中的消息条数阈值
#log.flush.interval.messages=10000# 消息刷新到磁盘中的最大时间间隔,1s
#log.flush.interval.ms=1000############################# Log Retention Policy ############################# # 日志保留小时数，超时会自动删除，默认为7天
log.retention.hours=168# 日志保留大小，超出大小会自动删除，默认为1G
#log.retention.bytes=1073741824# 日志分片策略，单个日志文件的大小最大为1G，超出后则创建一个新的日志文件
log.segment.bytes=1073741824# 每隔多长时间检测数据是否达到删除条件,300s
log.retention.check.interval.ms=300000############################# Zookeeper #############################
# Zookeeper连接信息，如果是zookeeper集群，则以逗号隔开
zookeeper.connect=10.0.0.161:2181,10.0.0.162:2181,10.0.0.163:2181# 连接zookeeper的超时时间,6s
zookeeper.connection.timeout.ms=60003.创建数据存储的目录
[root@kafka-node1 config]# mkdir ../data
[root@kafka-node1 opt]# scp -rp kafka_2.12-2.3.0 root@10.0.0.162:/opt
[root@kafka-node1 opt]# scp -rp kafka_2.12-2.3.0 root@10.0.0.163:/opt4.修改162和163的server.properties文件中的broker.id
node1
broker.id=1node2
broker.id=2node3
broker.id=35.启动kafka集群
[root@kafka-node2 bin]# export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"
[root@kafka-node2 bin]# ./kafka-server-start.sh ../config/server.properties #启动测试
[root@kafka-node2 bin]# ./kafka-server-start.sh -daemon ../config/server.properties #放入后台6.使用kafka创建一个topic
[root@yinwu-kafka-node1-172 bin]# ./kafka-topics.sh \
--create \
--zookeeper 10.0.0.161:2181,10.0.0.162:2181,10.0.0.163:2181 \
--partitions 3 \
--replication-factor 1 \
--topic yinwu-kafka7.消息发布者测试
[root@yinwu-kafka-node1-172 bin]# ./kafka-console-producer.sh \
--broker-list 10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092 \
--topic yinwu-kafka8.消息订阅者测试
[root@yinwu-kafka-node1-172 bin]# ./kafka-console-consumer.sh \
--bootstrap-server 10.0.0.161:9092,10.0.0.162:9092,10.0.0.163:9092 \
--topic yinwu-kafka \
--from-beginning