分布式消息队列--Kafka_kafka分布式消息队列

定义与特性

Kafka是一个支持分区，多副本，且基于zookeeper协调的分布式消息系统。多个分区副本中只有Leader副本提供读写服务，其他Follower副本被动复制Leader的结果且不提供读写服务，目的主要是为了保证多副本数据与消费的一致性。

特性：用Scala语言编写的可以实时处理大数据相关需求场景，如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎、访问日志、消息服务等。

使用场景

1. 日志收集：收集各个服务的log，通过以统一接口服务的方式开放给各种消费者，如hadoop, Hbase, Solr等
2. 消息系统：生产者消费者解耦、缓存消息等
3. 运营指标：用来记录运营监控数据，包括各种分布式应用的数据，生产各种操作的几种反馈，如报警和报告
4. 用户活动跟踪：用于记录web/app用户的各种活动

基本概念

基本使用

本地示例安装Kafka版本2.8.2，对应Scala版本2.12

环境准备

1. 安装jdk
   yum install java-1.0.0-openjdk* -y

2. 安装zookeeper
   wget https://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8.bin.tar.gz
   tar -zxvf apache-zookeeper-3.5.8-bin.tar.gz
   cd apache-zookeeper-3.5.8-bin
   cp conf/zoo_sample.cfg conf/zoo.cfg

3. 启动zookeeper
   bin/zkServer.sh start conf/zoo.cfg
   bin/zkCli.sh -server 192.168.126.135:2181
   ls /

单机部署

1. 下载安装包

cd /usr/local
#Scala版本2.12，Kafka版本2.8.2
wget https://mirrors.aliyun.com/apache/kafka/2.8.2/kafka_2.12-2.8.2.tgz
tar -zxf kafka_2.12-2.8.2.tgz
cd kafka_2.12-2.8.2/ && ll

2. 修改conf/server.properties配置

vi config/server.properties

#broker.id属性在Kafka集群中必须唯一
broker.id=0
#Kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.126.135:9092
#Kafka消息存储文件路径
log.dirs=/usr/local/kafka_2.12-2.8.2/kafka-logs
#Kafka连接zookeeper地址
zookeeper.connect=192.168.126.135:2181

:wq

4. 启动kafka
   kafka-server-start.sh [-daemon] server.properties
   (注意：启动kafka时会使用Linux主机名关联的IP地址，故需要把主机名和Linux的IP映射配置到本地host中，/etc/hosts)

启动命令：
bin/kafka-server-start.sh -daemon config/server.properties
或
bin/kafka-server-start.sh config/server.properties &
#查看zookeeper中kafka节点
bin/zkCli.sh
ls /brokers/ids

5. 停止Kafka

bin/kafka-server-stop.sh

基本使用

1. 主题操作

#创建一个分区数与备份因子均为1且名称为test的Topic (注意：若Producer发消息到某个指定的且不存在的Topic，则kafka会自动创建)
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic test --partitions 1 --replication-factor 1

#查看kafka中存在的topic列表
bin/kafka-topics.sh --list --zookeeper 192.168.126.135:2181

#删除主题
bin/kafka-topics.sh --delete --zookeeper 192.168.126.135:2181 --topic test

2. 发送与消费消息

#往指定kafka集群的指定topic发送消息
bin/kafka-console-producer.sh --broker-list 192.168.126.135:9092 --topic jeffrey

通过Offset Explorer工具查看kafka消息如下：

#从指定kafka集群的指定topic消费最新消息（默认）
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey

#从指定kafka集群的指定topic从头开始消费消息
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --from-beginning

#从指定kafka集群的指定多个topic消费消息[从头开始消费]
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --whitelist “jeffrey|qiutee” --from-beginning

3. 单播消息与多播消息

#单播消息(一个topic中的一个消息只能被同一个消费者组下的某一个消费者消费类似Queue模式，实现只需将所有消费者置于同一个消费者组即可)
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --consumer-property group.id=zfy-consumer-group

#多播消息(一个消息能被多个消费者消费类似发布订阅模式，实现只需将所有消费者置于不同消费者组即可)
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9092 --topic jeffrey --consumer-property group.id=wqt-consumer-group

4. 消费者组操作

#查看指定kafka集群中消费者组列表
bin/kafka-consumer-groups.sh --list --bootstrap-server 192.168.126.135:9092

#查看指定kafka集群的指定消费者组的描述信息（可查询消息消费情况）
bin/kafka-consumer-groups.sh --describe --bootstrap-server 192.168.126.135:9092 --group zfy-consumer-group
(topic:当前主题；partition:当前分区；current-offset:当前消费者组在当前主题当前分区中的已消费偏移量；log-end-offset:当前主题对应当前分区中消息的结束偏移量HW；lag:当前消费者组未消费的消费数，值为当前主题分区的LOG-END-OFFSET(LEO)减去当前消费者组已消费偏移量CURRENT-OFFSET)

主题与分区

Topic是消息分类的名称，一个Topic下可以有多个Partition日志文件。
Partition是一个有序消息序列，一个Partition对应一个Commit log文件，Partition中的offset唯一标识某个分区中的消息；一个Partition中的offset是唯一的，不同Partition的offset可能相同。
Kafka消息日志默认保留时间168h（近一周），可通过配置log.retention.hours自定义保留时间。
每个consumer是基于自己在commit log中的offset来进行消费的，消费offset由consumer自身维护；一般而言partition中的消息是按照顺序消费的，也可以通过offset重复消费或跳过消费。

#创建多个分区的主题
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic jack --partitions 2 --replication-factor 1

#查看topic描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic jack

topic描述信息介绍：

• leader节点负责给定partition的所有读写请求
• replicas 表示某个partition在哪几个broker上存在备份。不管这几个是不是”leader“，甚至这个节点挂了也会列出
• isr 是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点

#增加topic的分区数量（注意：topic的分区数量只能增加不能减少，减少topic分区会报错）
bin/kafka-topics.sh --alter --zookeeper 192.168.126.135:2181 --topic jack --partitions 3

为什么要对Topic下数据进行分区存储？

1、数据分布式存储，commit log文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，理论上一个topic可以处理任意数量的数据。
2、为了提高并行度。

集群部署

企业级Kafka集群一般是多个broker实例部署到多台不同的机器节点上，本次演示由于资源有限，将在同一台虚拟机上部署三个broker实例组成一个Kafka伪集群。

集群搭建步骤如下：

1. 环境依赖

依据上述 基本使用-环境准备 章节介绍，确保集群部署前安装完成jdk，安装与启动完成zookeeper

2. 下载安装Kafka

依据上述 单机部署-下载安装包章节介绍，下载解压Kafka安装包并进入解压后的目录

3. 创建kafka集群配置目录与日志存储目录

#创建kafka集群配置目录
mkdir -p /usr/local/kafka-cluster/node1/config
mkdir -p /usr/local/kafka-cluster/node2/config
mkdir -p /usr/local/kafka-cluster/node3/config
#创建kafka集群日志存储目录
mkdir -p /usr/local/kafka-cluster/node1/kafka-logs
mkdir -p /usr/local/kafka-cluster/node2/kafka-logs
mkdir -p /usr/local/kafka-cluster/node3/kafka-logs

3. 拷贝并修改配置文件

拷贝三份config/server.properties配置文件到kafka集群配置目录下并进行相应的配置修改

cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node1/config/
cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node2/config/
cp /usr/local/kafka_2.12-2.8.2/config/server.properties /usr/local/kafka-cluster/node3/config/

分别对node1~node3下的config/server.properties的配置文件进行相应修改（下面以node1节点配置为例）

vi /usr/local/kafka-cluster/node1/config/server.properties

#集群broker id（注意：确保集群中每个broker实例都是唯一的）
broker.id=1
#当前broker实例部署的机器IP和服务提供的端口
listeners=PLAINTEXT://192.168.126.135:9093
#当前broker实例节点的日志存储目录（包含消息日志、索引日志以及时间索引日志）
log.dirs=/usr/local/kafka-cluster/node1/kafka-logs
#zk连接地址（注意：若存在zk集群，则多个zk节点地址用逗号隔开即可）
zookeeper.connect=192.168.126.135:2181

:wq

4. 启动kafka集群

分别启动三个broker实例即可

bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node1/config/server.properties
bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node2/config/server.properties
bin/kafka-server-start.sh -daemon /usr/local/kafka-cluster/node3/config/server.properties

5. 验证集群

#查看zk中三个broker实例是否都已经注册完成
/usr/local/zookeeper-3.5.10/bin/zkCli.sh -server 192.168.126.135:2181
ls /brokers/ids

quit

6. 集群操作演示

#创建一个分区数为2备份因子为3的名称为test-cluster-topic的主题（备份因子为3表示该topic的单个分区的总副本数为3，包含分区一个leader副本和两个follower副本）
bin/kafka-topics.sh --create --zookeeper 192.168.126.135:2181 --topic test-cluster-topic --partitions 2 --replication-factor 3

#查看新创建的topic的描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic test-cluster-topic

注：图中第一行描述当前topic相关信息；下面每行表示该topic每个分区的信息，其中Leader表示当前分区的leader副本所在的broker节点，Replicas表示当前分区在集群中哪些broker节点存在备份（包含未存活的broker节点以及非leader副本所在的broker节点），Isr则表示当前集群中存活着的并且已同步备份了当前分区的broker节点；Leader副本所在broker节点负责处理消息的读写请求，而Follower副本对应broker则是接收Leader副本的broker发送过来的消息并对消息进行备份存储。

#发送消息
bin/kafka-console-producer.sh --broker-list 192.168.126.135:9093,192.168.126.135:9094,192.168.126.135:9095 --topic test-cluster-topic
#消费消费
bin/kafka-console-consumer.sh --bootstrap-server 192.168.126.135:9093,192.168.126.135:9094,192.168.126.135:9095 --topic test-cluster-topic --from-beginning

分区容错性测试：

由上图可知，主题test-cluster-topic分区0的Leader副本对应broker-1实例节点，为了演示分区容错，这里将分区0的Leader副本对应的broker节点进程杀掉，看是否会选举出新的broker节点作为分区0的Leader副本所对应的broker节点
ps -ef|grep kafka
kill 119259（这里我的broker-1实例进程id为119259）
等待broker-1进程杀掉后再次查看主题test-cluster-topic的描述信息
bin/kafka-topics.sh --describe --zookeeper 192.168.126.135:2181 --topic test-cluster-topic

可以看到，根据分区0 Isr中已存活且已同步备份的节点集来看，新选举出的主题test-cluster-topic分区0的Leader副本对应的实例节点为broker-3
（说明：一般情况同一主题下不同分区的Leader副本对应节点是不一样的，但在选举过程中，偶尔可能会出现同一主题下不同分区的Leader副本对应相同的broker节点）

查看zk中topic分区的state
get /brokers/topics/test-cluster-topic/partitions/0/state

（说明：Kafka相关集群信息保存在zk中，保持自身无状态，有利于水平扩容）

代码实战

producer

public class KfkProducer {
   
    private static final String CONNECT_STR = "192.168.126.129:9092";
    private static final String KFK_TOPIC = "test";
    private static final CountDownLatch CDL = new CountDownLatch(1);
    public static void main(String[
1
2
3
4
5