消息队列--Kafka

1. Kafka简介
2. 集群部署
3. 配置Kafka
4. 测试Kafka

1.Kafka简介

数据缓冲队列。同时提高了可扩展性。具有峰值处理能力，使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

Kafka是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、web/nginx日志、访问日志，消息服务等等，用scala语言编写，Linkedin于2010年贡献给了Apache基金会并成为顶级开源项目。

特性:

• 高吞吐量：kafka每秒可以处理几十万条消息。

• 可扩展性：kafka集群支持热扩展- 持久性、

• 可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失

• 容错性：允许集群中节点失败（若副本数量为n,则允许n-1个节点失败）

• 高并发：支持数千个客户端同时读写

它主要包括以下组件：

话题（Topic）：是特定类型的消息流。(每条发布到 kafka 集群的消息属于的类别，即 kafka 是面向 topic 的。)
生产者（Producer）：是能够发布消息到话题的任何对象(发布消息到 kafka 集群的终端或服务).
消费者（Consumer）：可以订阅一个或多个话题，从而消费这些已发布的消息。
服务代理（Broker）：已发布的消息保存在一组服务器中，它们被称为代理（Broker）或Kafka集群。

partition（区）：每个 topic 包含一个或多个 partition。
replication：partition 的副本，保障 partition 的高可用。
leader：replica 中的一个角色， producer 和 consumer 只跟 leader 交互。
follower：replica 中的一个角色，从 leader 中复制数据。
zookeeper：kafka 通过 zookeeper 来存储集群的信息。

Zookeeper:

ZooKeeper是一个分布式协调服务，它的主要作用是为分布式系统提供一致性服务，提供的功能包括：配置维护、分布式同步等。Kafka的运行依赖ZooKeeper。  也是java微服务里面使用的一个注册中心服务

ZooKeeper主要用来协调Kafka的各个broker，不仅可以实现broker的负载均衡，而且当增加了broker或者某个broker故障了，ZooKeeper将会通知生产者和消费者，这样可以保证整个系统正常运转。

在Kafka中,一个topic会被分成多个区并被分到多个broker上，分区的信息以及broker的分布情况与消费者当前消费的状态信息都会保存在ZooKeeper中。

2.集群部署

        2.1环境

系统：Centos-Stream7

节点：

192.168.26.166   es01 ​

192.168.26.170   es02 ​

192.168.26.171   es03

软件版本：kafka_2.12-3.0.2.tgz

        2.2  安装配置jdk8

#yum install -y java-1.8.0-openjdk

        2.3  安装配置zookeeper

在配置中要注意每个配置项后面不要有空格否则会导致zookeeper启动不起来！！！！

Kafka运行依赖ZK，Kafka官网提供的tar包中，已经包含了ZK，这里不再额外下载ZK程序。

配置相互解析---三台机器(在es集群上安装的kafka)：

# vim /etc/hosts

192.168.26.166   es01 ​

192.168.26.170   es02 ​

192.168.26.171   es03

安装Kafka：

# wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/3.0.2/kafka_2.12-3.0.2.tgz

# tar xzvf kafka_2.12-2.8.0.tgz -C /usr/local/

# mv /usr/local/kafka_2.12-2.8.0/ /usr/local/kafka/

配置zookeeper：

在es01节点中：

# sed -i 's/^[^#]/#&/' /usr/local/kafka/config/zookeeper.properties

# vim /usr/local/kafka/config/zookeeper.properties  #添加如下配置
dataDir=/opt/data/zookeeper/data  # 需要创建，所有节点一致
dataLogDir=/opt/data/zookeeper/logs # 需要创建，所有节点一致
clientPort=2181 
tickTime=2000 
initLimit=20 
syncLimit=10 

# 以下 IP 信息根据自己服务器的 IP 进行修改
server.1=192.168.19.20:2888:3888  //kafka集群IP:Port
server.2=192.168.19.21:2888:3888
server.3=192.168.19.22:2888:3888

#创建data、log目录

# mkdir -p /opt/data/zookeeper/{data,logs}

#创建myid文件

# echo 1 > /opt/data/zookeeper/data/myid     #myid号按顺序排

在es02节点中：

# sed -i 's/^[^#]/#&/' /usr/local/kafka/config/zookeeper.properties

# vim /usr/local/kafka/config/zookeeper.properties
dataDir=/opt/data/zookeeper/data 
dataLogDir=/opt/data/zookeeper/logs
clientPort=2181 
tickTime=2000 
initLimit=20 
syncLimit=10 
server.1=192.168.19.20:2888:3888
server.2=192.168.19.21:2888:3888
server.3=192.168.19.22:2888:3888

#创建data、log目录

# mkdir -p /opt/data/zookeeper/{data,logs}

#创建myid文件

# echo 2 > /opt/data/zookeeper/data/myid

在es03节点中：

# sed -i 's/^[^#]/#&/' /usr/local/kafka/config/zookeeper.properties

# vim /usr/local/kafka/config/zookeeper.properties
dataDir=/opt/data/zookeeper/data 
dataLogDir=/opt/data/zookeeper/logs
clientPort=2181 
tickTime=2000 
initLimit=20 
syncLimit=10 
server.1=192.168.19.20:2888:3888
server.2=192.168.19.21:2888:3888
server.3=192.168.19.22:2888:3888

#创建data、log目录

# mkdir -p /opt/data/zookeeper/{data,logs}

#创建myid文件

# echo 3 > /opt/data/zookeeper/data/myid

配置项含义：

dataDir     ZK数据存放目录。
dataLogDir  ZK日志存放目录。
clientPort  客户端连接ZK服务的端口。
tickTime    ZK服务器之间或客户端与服务器之间维持心跳的时间间隔。
initLimit   允许follower连接并同步到Leader的初始化连接时间，当初始化连接时间超过该值，则表示连接失败。
syncLimit   Leader与Follower之间发送消息时如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃。
server.1=192.168.19.20:2888:3888    2888是follower与leader交换信息的端口，3888是当leader挂了时用来执行选举时服务器相互通信的端口。

3.配置Kafka

        3.1  配置

# sed -i 's/^[^#]/#&/' /usr/local/kafka/config/server.properties

# vim /usr/local/kafka/config/server.properties  #在最后添加

broker.id=1  #改 
listeners=PLAINTEXT://192.168.19.20:9092   #改
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/data/kafka/logs  
num.partitions=6
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=2
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=536870912
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.19.20:2181,192.168.19.21:2181,192.168.19.22:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

[root@es01 ~]# mkdir -p /opt/data/kafka/logs

        3.2  其他节点配置

只需把配置好的安装包直接分发到其他节点，修改 Kafka的broker.id和 listeners就可以了。

        3.3  配置项含义

broker.id 
    每一个broker在集群中的唯一标识，要求是正数。在改变IP地址，不改变broker.id的时不会影响consumers
listeners=PLAINTEXT://192.168.19.22:9092       
    监听地址
num.network.threads  
    broker 处理消息的最大线程数，一般情况下不需要去修改
num.io.threads
    broker处理磁盘IO 的线程数 ，数值应该大于你的硬盘数
socket.send.buffer.bytes          
    socket的发送缓冲区
socket.receive.buffer.bytes        
    socket的接收缓冲区
socket.request.max.bytes
    socket请求的最大数值，防止serverOOM，message.max.bytes必然要小于socket.request.max.bytes，会被topic创建时的指定参数覆盖
log.dirs        日志文件目录
num.partitions
num.recovery.threads.per.data.dir   每个数据目录(数据目录即指的是上述log.dirs配置的目录路径)用于日志恢复启动和关闭时的线程数量。
offsets.topic.replication.factor

transaction state log replication factor  事务主题的复制因子（设置更高以确保可用性）。 内部主题创建将失败，直到群集大小满足此复制因素要求

log.cleanup.policy = delete
    日志清理策略 选择有：delete和compact 主要针对过期数据的处理，或是日志文件达到限制的额度，会被 topic创建时的指定参数覆盖
log.cleanup.interval.mins=1
    指定日志每隔多久检查看是否可以被删除，默认1分钟    
log.retention.hours
    数据存储的最大时间 超过这个时间 会根据log.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据。log.retention.bytes和log.retention.minutes或者log.retention.hours任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖

log.segment.bytes
    topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖
log.retention.check.interval.ms 
    文件大小检查的周期时间，是否触发 log.cleanup.policy中设置的策略
zookeeper.connect   
    ZK主机地址，如果zookeeper是集群则以逗号隔开。
zookeeper.connection.timeout.ms     
    连接到Zookeeper的超时时间。

4.测试Kafka

        4.1  启动zookeeper集群

在三个节点依次执行：

# cd /usr/local/kafka

# nohup bin/zookeeper-server-start.sh config/zookeeper.properties &

查看端口：

# netstat -lntp | grep 2181

        4.2  启动Kafka

在三个节点依次执行：

# cd /usr/local/kafka

# nohup bin/kafka-server-start.sh config/server.properties &

        4.3  测验

验证  在192.168.26.166上创建topic：

# bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic

参数解释：

–zookeeper指定zookeeper的地址和端口，
–partitions指定partition的数量，
–replication-factor指定数据副本的数量

在26.170上面查询192.168.26.166上的topic：

[root@es03 kafka]# bin/kafka-topics.sh --zookeeper 192.168.26.166:2181 --list
testtopic

        （二）模拟消息生产和消费

发送消息到192.168.26.166：

[root@es01 kafka]# bin/kafka-console-producer.sh --broker-list 192.168.19.20:9092 --topic testtopic

>hello
>你好呀
>

从192.168.26.171接受消息：

[root@es02 kafka]# bin/kafka-console-consumer.sh --bootstrap-server  192.168.19.21:9092 --topic testtopic --from-beginning