65、zookeeper+kafka实现订阅号生产及阅读

前言回顾

elk

es–elasticsearch

l–logstash

k–kibana

f–filebeat

一、elkfk

kafka带入集群当中

zookeeper集群+kefka集群

zookeeper是一个开源的，分布式的，为分布式架构提供协调服务的APACHE的项目。

保存元数据。

1.1、zookeeper的工作机制：

观察者模式设计的分布式服务器管理架构。

负责存储和管理元数据，记录集群的变化。保存集群变化的信息。

1.2、zookeeper

1.2.1、zookeeper的特点：

1、在集群中分为领导者和追随者，组成的集群。

2、只要有半数以上的节点正常工作，整个zookeeper就可以正常工作。zookeeper在部署时一般选择奇数台。

3、全局的数据一致，每个zookeeper不论领导者还是追随者，在访问他们的数据都是一致的。

4、数据更新的原子性，一次更新数据，要么都成功，要么都失败。

5、数据更新的实时性。

6、领导者追随者根据投票产生。

1.2.2、选举机制：

A B C

1、服务器A启动，发起一次选举，A会投自己一票。A有一票，不够半数。选举无法完成，A进行lookig。

2、服务器B启动，再发起一次选举，服务B也投在自己一票，服务器A和B会做个比较，myid，谁的myid大。

如果A比B小，A会把票改投给B，2票，B自动当选为leader。

3、C启动，自动成为追随者，A也会成为追随者。

二、zookeeper集群安装部署

zookeeper1 kafka1 192.168.168.11

zookeeper2 kafka2 192.168.168.12

zookeeper3 kafka3 192.168.168.13

zookeeper保存kafka集群的信息，记录服务器的ip地址。

在这里插入图片描述

-----------------------三台机器-----------------

systemctl stop firewalld

setenforce 0

yum -y install ntpdate  ##安装时间同步工具

ntpdate ntp-aliyun.com

yum install -y java-1.8.0-openjdk java-1.8.0-openjdk-devel

java -version

安装 Zookeeper
cd /opt
tar -zxvf apache-zookeeper-3.5.7-bin.tar.gz
mv apache-zookeeper-3.5.7-bin /opt/zookeeper

//修改配置文件
cd /opt/zookeeper/conf/
cp zoo_sample.cfg zoo.cfg

vim zoo.cfg
tickTime=2000   #通信心跳时间，Zookeeper服务器与客户端心跳时间，单位毫秒
initLimit=10    #Leader和Follower初始连接时能容忍的最多心跳数（tickTime的数量），这里表示为10*2s
syncLimit=5     #Leader和Follower之间同步通信的超时时间，这里表示如果超过5*2s，Leader认为Follwer死掉，
                 并从服务器列表中删除Follwer
dataDir=/opt/zookeeper/data      ●修改，指定保存Zookeeper中的数据的目录，目录需要单独创建
dataLogDir=/opt/zookeeper/logs   ●添加，指定存放日志的目录，目录需要单独创建
clientPort=2181   #客户端连接端口
#添加集群信息
server.1=192.168.168.11:3188:3288
server.2=192.168.168.12:3188:3288
server.3=192.168.168.13:3188:3288

-------------------------------------------------------------------------------------

server.A=B:C:D
●A是一个数字，表示这个是第几号服务器。集群模式下需要在zoo.cfg中dataDir指定的目录下创建一个文件myid，
这个文件里面有一个数据就是A的值，Zookeeper启动时读取此文件，
拿到里面的数据与zoo.cfg里面的配置信息比较从而判断到底是哪个server。

●B是这个服务器的地址。

●C是这个服务器Follower与集群中的Leader服务器交换信息的端口。

●D是万一集群中的Leader服务器挂了，需要一个端口来重新进行选举，选出一个新的Leader，

#server.1数字id也就是服务器对应的myid

#192.168.233.21 服务器的ip地址

#3188：zookeeper集群内部通信的端口

#3288：重新选举端口，万一leader挂了，用这个端口进行内部通信，选举新的leader。

#而这个端口就是用来执行选举时服务器相互通信的端口。

//拷贝配置好的 Zookeeper 配置文件到其他机器上
scp /opt/zookeeper/conf/zoo.cfg 192.168.168.12:/opt/zookeeper/conf/
scp /opt/zookeeper/conf/zoo.cfg 192.168.168.13:/opt/zookeeper/conf/

//在每个节点上创建数据目录和日志目录
mkdir /opt/zookeeper/data
mkdir /opt/zookeeper/logs

//在每个节点的dataDir指定的目录下创建一个 myid 的文件
echo 1 > /opt/zookeeper/data/myid
echo 2 > /opt/zookeeper/data/myid
echo 3 > /opt/zookeeper/data/myid

//配置 Zookeeper 启动脚本
vim /etc/init.d/zookeeper
#!/bin/bash
#chkconfig:2345 20 90
#description:Zookeeper Service Control Script
ZK_HOME='/opt/zookeeper'
case $1 in
start)
	echo "---------- zookeeper 启动 ------------"
	$ZK_HOME/bin/zkServer.sh start
;;
stop)
	echo "---------- zookeeper 停止 ------------"
	$ZK_HOME/bin/zkServer.sh stop
;;
restart)
	echo "---------- zookeeper 重启 ------------"
	$ZK_HOME/bin/zkServer.sh restart
;;
status)
	echo "---------- zookeeper 状态 -----



//	设置开机自启
chmod +x /etc/init.d/zookeeper
chkconfig --add zookeeper

//分别启动 Zookeeper
service zookeeper start

//查看当前状态
service zookeeper status


[root@mysql1 config]# service zookeeper status
---------- zookeeper 状态 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower  ##追随者



[root@mysql2 config]#  service zookeeper status
---------- zookeeper 状态 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: follower   ##追随者



[root@mysql3 config]# service zookeeper status
---------- zookeeper 状态 ------------
/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Client port found: 2181. Client address: localhost.
Mode: leader  ##领导者

三、kafka

3.1、kafka概述

3.1.1、消息队列：MQ

在高并发的情况下，同步请求来不及处理，请求太多会造成阻塞。
比如说大量请求并发到数据库，too many connection 报错。
消息队列，使用异步处理方式，可以缓解系统处理请求的压力。

例如：短信验证码、邮箱验证都是消息队列的一种

在这里插入图片描述

3.2、kafka的作用

1、异步处理

2、系统解耦

每个系统之间独立运行，互相之间没有必然的依赖关系。

微服务架构中的通信对于解耦来说至关重要。

各个微服务之间独立运行，分别处理各自的请求和消息。提高整个系统的吞吐量和处理能力。

电商的订单系统，网站的工单系统，典型的一个消息队列场景。

3、负载均衡

消息队列的负载均衡：把任务发送到多个生产者，多个生产者可以并行处理队列中的消息

在这里插入图片描述

谁发谁处理

4、流量控制和限流：

通过延迟方法，处理生产速率和消费者的处理速度（代码控制）。

5、数据同步和数据分发(重点掌握)

可以实现跨系统的数据同步和日志收集。

6、任务调度和定时任务

7、实时数据处理

8、备份和恢复

3.3、消息队列的模式:

1、点对点一对一消费者消费完数据之后，生产者会自动清除已消费的数据。

一个生产者对应一个消费者（淘汰）

2、发布/订阅模式（一对多，观察者模式，消费者数据在消费完之后不会被清除（保留一段时间））生产者发布一个消息，可以是一个消费者使用，也可以是多个消费者同时使用。（主流）

kafka就是发布/订阅模式的消息队列。RAbbitMQ也是发布/订阅的消息队列。小集群内部使用。

大数据的实时处理领域。

3.4、kafka的特性：

高吞吐量，低延迟。

每秒可以处理几十万数据。延迟只有几毫秒。

集群的可扩展性（支持热扩展）

消息的持久化：生产者发布的消息可以保存到磁盘当中，防止数据丢失。（有时间限制）

容错性：挂了一个可以继续使用

高并发：数千个客户端可以同时读写。

3.5、kafka的组件：

topic主题，kafka的基本单元，所有生产者发布的消息都是发到主题。

消费者订阅主题，然后消费生产者发布的消息。

生产者：生产者把消息发布到主题

消费者：订阅主题，消费生产者发布的消息

分区：每个主题都可以分成多个分区。每个分区都是数据的有序子集。

分区当中保留数据，按照偏移量来有序的存储数据，消费者可以根据偏移量来消费指定分区当中的消息（一般不用）

在这里插入图片描述

偏移量：消息在分区当中的唯一标识，跟踪和定位消息所在的位置。消费可以根据偏移量来处理信息。

分区还有备份的作用：我们在创建主题时创建主题，创建分区时要指定副本数。

分区和我们执行的集群机器数量一般是保持一致的。

副本：备份分区中的消息。最少要2，互为备份。

经纪人 broker 经纪人处理生产者和消费者的请求（kafka）元数据（zookeeper）

zookeeper：保存元数据-------ip地址。

四、kafka的工作流程(面试)、安装部署：

4.0、kafka的工作流程

生产者将消息发布到指定的主机，每个消息都附带一个key和value

主题是有多个分区的，生产者把消息写入一个分区（带偏移量）

经纪人（kafka）分配和处理生产者的发布请求，偏移量也是经纪人分配（在分区中是唯一的）。

消费者订阅主题，获取全量的消费者的消费信息（默认模式），也可以从执行的分区获取消息（代码来完成，一边拿不用）

生产者发布的消息会在本地保留一段时间，防止消费者有延迟或者处理速度过慢，导致没有成功消费。保留时间：7天（默认）。

4.1、kafka安装部署

--------------------------kafka安装部署---------------------------

tar zxvf kafka_2.13-3.4.1.tgz
mv kafka_2.13-3.4.1 /usr/local/kafka

//修改配置文件
cd /usr/local/kafka/config/
cp server.properties{,.bak}

vim server.properties
broker.id=0    ●21行，broker的全局唯一编号，每个broker不能重复，因此要在其他机器上配置 **broker.id=1、broker.id=2**
**listeners=PLAINTEXT://192.168.168.11:9092**   ●31行，指定监听的IP和端口，如果修改每个broker的IP需区分开来，也可保持默认配置不用修改
num.network.threads=3    #42行，broker 处理网络请求的线程数量，一般情况下不需要去修改
num.io.threads=8         #45行，用来处理磁盘IO的线程数量，数值应该大于硬盘数
socket.send.buffer.bytes=102400       #48行，发送套接字的缓冲区大小
socket.receive.buffer.bytes=102400    #51行，接收套接字的缓冲区大小
socket.request.max.bytes=104857600    #54行，请求套接字的缓冲区大小
**log.dirs=/usr/local/kafka/logs**        #60行，kafka运行日志存放的路径，也是数据存放的路径
num.partitions=1    #65行，topic在当前broker上的默认分区个数，会被topic创建时的指定参数覆盖
num.recovery.threads.per.data.dir=1    #69行，用来恢复和清理data下数据的线程数量
log.retention.hours=168    #103行，segment文件（数据文件）保留的最长时间，单位为小时，默认为7天，超时将被删除
log.segment.bytes=1073741824    #110行，一个segment文件最大的大小，默认为 1G，超出将新建一个新的segment文件
**zookeeper.connect=192.168.168.11:2181,192.168.168.12:2181,192.168.168.13:2181**    ●123行，配置连接Zookeeper集群地址

//修改环境变量日志段是主题分区日志文件的一部分。
vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin


source /etc/profile

//配置 kafka 启动脚本
vim /etc/init.d/kafka
#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
	echo "---------- Kafka 启动 ------------"
	${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
	echo "---------- Kafka 停止 ------------"
	${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
	$0 stop
	$0 start
;;
status)
	echo "---------- Kafka 状态 ------------"
	count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
	if [ "$count" -eq 0 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac

//设置开机自启
chmod +x /etc/init.d/kafka
chkconfig --add kafka

//分别启动 Kafka
service kafka start

kafka的默认端口是9092

netstat -antp | grep 9092

[root@mysql1 config]# netstat -antp | grep 9092
tcp6       0      0 192.168.168.11:9092     :::*                    LISTEN      20187/java

4.2、测试zookeeper+kafka工作情况

创建主题

[root@mysql1 config]# kafka-topics.sh --create --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --replication-factor 2 --partitions 3 --topic test1
Created topic test1.


Created topic test1.

--replication-factor 2 创建分区的副本数2

partitions 3 分区数3

--topic test1 指定主题的名称

生产者生产消息命令

[root@mysql1 config]# kafka-console-producer.sh --broker-list 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --topic test1
>1
>3
>4
>34
>43
>
查看消费者消费生产者生产的消息
[root@mysql2 config]# kafka-console-consumer.sh --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --topic test1 --from-beginning 
1
3
4
34
43

[root@mysql3 config]# kafka-console-consumer.sh --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --topic test1 --from-beginning 
1
3
4
34
43


查看消费者列表
[root@mysql2 config]# kafka-topics.sh --list --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
__consumer_offsets
test1


查看当前主题详情
[root@mysql3 config]# kafka-topics.sh --describe --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
Topic: test1	TopicId: idxxP4JMR_uejNmJTFvraA	PartitionCount: 3	ReplicationFactor: 2	Configs: 
	Topic: test1	Partition: 0	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test1	Partition: 1	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test1	Partition: 2	Leader: 1	Replicas: 1,2	Isr: 1,2


查看当前主题详情

topic：主题名称

partition：分区 偏移量

leader：分区的领导者，用来处理分区的读写操作。只有在指定写分区和只当读分区时才工作。如果不是指定，全量展示，无意义。

replicas：副本，0、1、2---------broker id

ISR：表示与当前与领导者同步的副本。0 1 3

删除主题
kafka-topics.sh --delete --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --topic test2


[root@mysql3 config]# kafka-topics.sh --list --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
__consumer_offsets
test1
test2
[root@mysql3 config]# kafka-topics.sh --delete --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --topic test2
[root@mysql3 config]# kafka-topics.sh --list --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
__consumer_offsets
test1

五、课后总结：zookeeper就是保存集群的元数据

kafka的工作流程

组件的作用

kafka的消息堆积该如何解决：

主要在于消费者出现了延迟或者处理能力太差，导致消息堆积

1、减少kafka持久化的保存时间。

2、修改主题的分区数，扩大分区的数量，提高消费者获取的通道。##处理消息最有效的方法。

3、可以指定多个消费者共同工作，处理消息的积压。

[root@mysql3 config]# kafka-topics.sh --describe --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
Topic: test1	TopicId: idxxP4JMR_uejNmJTFvraA	PartitionCount: 3	ReplicationFactor: 2	Configs: 
	Topic: test1	Partition: 0	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test1	Partition: 1	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test1	Partition: 2	Leader: 1	Replicas: 1,2	Isr: 1,2

kafka-topics.sh --create --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --replication-factor 2 --partitions 3 --topic test1  
Created topic test1.

--replication-factor 2 创建分区的副本数2

partitions 3 分区数3

--topic test1 指定主题的名称

##扩大分区数量
kafka-topics.sh --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --alter --topic test1 --partitions 6
	

[root@mysql3 config]# kafka-topics.sh --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092 --alter --topic test1 --partitions 6
[root@mysql3 config]# kafka-topics.sh --describe --bootstrap-server 192.168.168.11:9092,192.168.168.12:9092,192.168.168.13:9092
Topic: test1	TopicId: idxxP4JMR_uejNmJTFvraA	PartitionCount: 6	ReplicationFactor: 2	Configs: 
	Topic: test1	Partition: 0	Leader: 0	Replicas: 0,1	Isr: 0,1
	Topic: test1	Partition: 1	Leader: 2	Replicas: 2,0	Isr: 2,0
	Topic: test1	Partition: 2	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test1	Partition: 3	Leader: 0	Replicas: 0,2	Isr: 0,2
	Topic: test1	Partition: 4	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: test1	Partition: 5	Leader: 2	Replicas: 2,1	Isr: 2,1