分布式消息队列之Kafka

简介总结

---

最重要的作用：解耦、削峰、异步处理**

---

1、 解耦：各个程序之间的一个耦合度，把kafka放到中间就是起到了解耦的作用
2、 削峰：当你的请求量到达封顶的状态，kafka就可以削弱请求的高峰，让吞吐量请求保持可控
3、异步：客户端写请求，并不是马上写到数据库当中。而是放到fakfa集群当中，等数据库有空闲资源的时候，再来拿取数据

总结：可以缓解后端数据库的压力，所有的数据暂存到本地。等后端的数据库有多余的空间再来取数据

优点：
1、分布式、支持分区、多副本
2、基于zookeeper协调分布式消息系统
3、特性，可以实时的处理大量的数据，满足各种场景的需求

概念：Producers(生产者)、Consumers(消费者)

理解：
1、生产者产生数据：-----web服务器产生数据，放到topic（主题）
2、消费者消费数据：-----数据库消费数据，拿取并存储

消费者组的概念：
1、这个消费者组当中有很多数据库，这个几个数据库，负责的主题的分区也不一样。

2、每个topic主题里面的每个分区对应一个消费者（负载均衡）

3、这些数据也有超时时间，如果过了超时时间数据就会被删掉。

4、而且消费者消费完数据，会被删除数据可以设置期限（例：两天，如果没被消费就删除）

`分区相当于磁盘，如果消费之都去同一个分区取数据，磁盘会承受不住

那么消费者如何去找到对应的分区呢？
1、结合zookeeper来使用，消费者需要先去找zookeeper，然后zookeeper告诉消费者对一个的一个分区（数据位置）。

2、如果当中kafka集群当中坏掉一台，zookeeper会告诉其他几台相应的信息。下次存数据的时候，不会去坏的那台去存数据了

3、所有的信息都会存放到zookeeper当中（类似：元数据）

Kafka集群的概念
1、Broker—kafka中的某台主机
2、Topic ----再主机中可以有多个主题，分配不同的场景
3、分区 — 一个主题可以有多个分区。例：三个分区，每个分区都会存放主题数据的三分之一
4、副本 — 一个分区可以有多个副本。保证数据的安全性

不同的主题放到不同的topic里面，一个主题可以存放多个分区、分区下还有副本 （保证数据的安全性）

每个分区存放在不同的主机上，减少对数据的损失

KafKa使用场景

• 日志收集：一个公司可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
• 消息系统：解耦和生产者和消费者、缓存消息等。
• 用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
• 运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
• 流式处理：比如spark streaming和storm
• 事件源

部署kafka集群

前提部署ZooKeePer集群，去我的上个文章部署 然后在每个zookeeper节点安装Kafka，同步部署

网盘地址：拿取kafka软件
链接：https://pan.baidu.com/s/1Q0ZRIp_-imKGStHXu13DnQ
提取码：tvqb

上传kafka软件kafka_2.12-2.2.0.tgz

1、解压

[root@centos1 ~]# tar xf kafka_2.12-2.2.0.tgz -C /usr/local/
[root@centos1 ~]# cd /usr/local/kafka_2.12-2.2.0/

2、修改配置文件

[root@centos1 kafka_2.12-2.2.0]# vim config/server.properties   //修改配置文件
#broker的全局唯一编号，不能重复
broker.id=0
#监听
listeners=PLAINTEXT://:9092  #开启此项
#日志目录
log.dirs=/data/kafka/log      #修改日志目录
#配置zookeeper的连接（如果不是本机，需要该为ip或主机名）
zookeeper.connect=localhost:2181
例：zookeeper.connect=192.168.1.10:2181,192.168.1.20:2181,192.168.1.30:2181
建议都写上，实现高可用

3、创建与配置文件相应的目录

[root@centos1 kafka_2.12-2.2.0]# mkdir -p /data/kafka/log

4、给kafka配置变量（软连接）

[root@centos1 ~]# vi /etc/profile
export KAFKA_HOME=/usr/local/kafka_2.12-2.2.0
export PATH=$KAFKA_HOME/bin:$PATH

[root@centos1 ~]# source /etc/profile

---

启动、停止、测试，kafka的使用方法

前台启动

[root@cong11 ~]# kafka-server-start.sh $KAFKA_HOME/config/server.properties

看到[Kafka Server 0], started (kafka.server.KafkaServer)，说明kafka服务启动成功

后台启动
[root@cong11 ~]# kafka-server-start.sh -daemon /usr/local/kafka_2.12-2.2.0/config/server.properties

关闭kafka

[root@cong11 ~]# jps  #查看id号
[root@cong11 ~]# kill -9 15550
或者：
[root@cong11 ~]# kafka-server-stop.sh  

测试是否启动

输入jps查看进程，如果可以看到Kafka进程，表示启动成功

[root@cong11 ~]# jps
2742 Kafka
3111 Jps
1711 QuorumPeerMain

[root@cong11 ~]# jps -m
13506 Jps -m
3507 Kafka /usr/local/kafka_2.12-2.2.0/config/server.properties
1711 QuorumPeerMain /usr/local/apache-zookeeper-3.5.5-bin/bin/../conf/zoo.cfg

可以看到我们的kafka和zookeeper的java进程

---

测试联通性

创建一个topic，测试是否同步到其他的zookeeper当中
创建一个副本 ---- 一个主题

[root@centos3 ~]# kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

到其他服务器查看，是否同步

[root@centos1 ~]# kafka-topics.sh --list --zookeeper localhost:2181 
Test    可以查看到其他服务器写的主题

启动生产着和消费着进行测试
生产者生产数据，消费着进行接收

第一台
启动生产者

[root@centos1 ~]# kafka-console-producer.sh --broker-list localhost:9092 --topic test

第三台
启动消费者

[root@centos3 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

--from-beginning的意思是，从头开始看生产着的信息。如果不加，就看不到之前的数据

在第一台输入，到消费者看是否看到
生产者

消费者

---

测试高可用

将生产者关机，看消费之是否可以查看到信息

[root@centos3 ~]# kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
Hellod    是可以的，因为有一个分区还有一个副本

日志

[root@centos3 ~]# ls /data/kafka/log/

查看主题的状态

[root@centos3 ~]# kafka-topics.sh --describe --zookeeper localhost:2181 --topic test

发现leader切换到第3个节点，第二个节点在活跃状态中消失