大数据之-Kafka_kafka注册中心

1Kafka定义

Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。
消息队列解决的问题是同步问题

1.1消息队列的作用

1） 解耦
允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。
2） 可恢复性
系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所
以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
3） 缓冲有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。
4） 灵活性 & 峰值处理能力
在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列
能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。
5） 异步通信
同步通信存在问题：造成系统开销响应时间比较大，在同步的环境每个服务执行完成，整个服务调用链才执行完成，如果由于网络存在一些问题，整个执行流程会受到影响。
很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。
分布式事务保证最终一致性。
异步的方式，可以使上游快速成功，极大提高了系统的吞吐量

1.2消息队列的两种模式

（1） 点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）
消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。
Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。

（2） 发布/订阅模式（一对多，消费者消费数据之后不会清除消息）
消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消
息。
发布订阅两种模式：
1、消费者主动拉取数据
2、队列主动推送数据----消费者消费速率不对等，可能会产生服务资源浪费，和服务崩溃。

1.2消息队列流派

1. 有Broker的MQ
   1、重topic -以top为单位 进行中转
   2、轻topic -支持多种模式
2. 无Broker的MQ
   1、ZeroMQ
   kafka是性能最快的消息中间件

2Kafka

2.1 Kafka架构

1. 生产者消费消息
2. Kafks集群管理消息
   1、kafka集群是运行在服务器的进程。
   2、Kafka集群会把消息按不同主题分区。
   3、Kafka会把数据做冗余备份(不在同一台服务器)，Leader,Follower(数据备份)
3. 消费者消费消息
   1、消费者组，某一个分区只能被同一个消费者组里面的消费者消费。
   消费者组，提高消费能力。
4. Zookeeper注册消息
   1、Kafka注册中心
   2、消费者消费信息保存在zk里，便于消费者挂掉之后，从zk中读取消费信息。
   0.9版本之前存在zk中，0.9之后保存在Kafka某个主题中。存磁盘(默认保存7天)

2.1 docker 部署Kafka集群

2.1.1 docker-compose安装

1、sudo curl -L “https://github.com/docker/compose/releases/download/1.28.2/docker-compose-$(uname-s)-$(uname -m)” -o /usr/local/bin/docker-compose
2、chmod +x /usr/local/bin/docker-compose

2.1.2 Kafka集群安装

1、创建虚拟网络
docker network create --driver bridge --subnet 172.23.0.0/16 --gateway 172.23.0.1 zoo_kafka
docker-compose.yml内容如下:

version: '2'
services:
 zoo1:
  image: zookeeper:3.4 # 镜像名称
  restart: always # 当发生错误时自动重启
  hostname: zoo1
  container_name: zoo1
  privileged: true
  ports: # 端口
   - 2181:2181
  volumes: # 挂载数据卷
   - ./zoo1/data:/data
   - ./zoo1/datalog:/datalog 
  environment:
   TZ: Asia/Shanghai
   ZOO_MY_ID: 1 # 节点ID
   ZOO_PORT: 2181 # zookeeper端口号
   ZOO_SERVERS: server.1=zoo1:2888:3888 server.2=zoo2:2888:3888 server.3=zoo3:2888:3888 # zookeeper节点列表
  networks:
   default:
    ipv4_address: 172.23.0.11

 zoo2:
  image: zookeeper:3.4
  restart: always
  hostname: zoo2
  container_name: zoo2
  privileged: true
  ports:
   - 2182:2181
  volumes:
   - ./zoo2/data:/data
   - ./zoo2/datalog:/datalog
  environment:
   TZ: Asia/Shanghai
   ZOO_MY_ID: 2
   ZOO_PORT: 2181
   ZOO_SERVERS: server.1=zoo1:2888:3888 server.2=zoo2:2888:3888 server.3=zoo3:2888:3888
  networks:
   default:
    ipv4_address: 172.23.0.12

 zoo3:
  image: zookeeper:3.4
  restart: always
  hostname: zoo3
  container_name: zoo3
  privileged: true
  ports:
   - 2183:2181
  volumes:
   - ./zoo3/data:/data
   - ./zoo3/datalog:/datalog
  environment:
   TZ: Asia/Shanghai
   ZOO_MY_ID: 3
   ZOO_PORT: 2181
   ZOO_SERVERS: server.1=zoo1:2888:3888 server.2=zoo2:2888:3888 server.3=zoo3:2888:3888
  networks:
   default:
    ipv4_address: 172.23.0.13

networks:
 default:
  external:
   name: zoo_kafka
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

Kafka：docker-compose.yml内容：
docker-compose up -d 启动

version: '2'

services:
 broker1:
  image: wurstmeister/kafka
  restart: always
  hostname: broker1
  container_name: broker1
  privileged: true
  ports:
   - "9091:9092"
  environment:
   KAFKA_BROKER_ID: 1
   KAFKA_LISTENERS: PLAINTEXT://broker1:9092
   KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker1:9092
   KAFKA_ADVERTISED_HOST_NAME: broker1
   KAFKA_ADVERTISED_PORT: 9092
   KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181
   #JMX_PORT: 9987
  volumes:
   - /var/run/docker.sock:/var/run/docker.sock
   - ./broker1:/kafka/kafka\-logs\-broker1
  external_links:
  - zoo1
  - zoo2
  - zoo3
  networks:
   default:
    ipv4_address: 172.23.0.14

 broker2:
  image: wurstmeister/kafka
  restart: always
  hostname: broker2
  container_name: broker2
  privileged: true
  ports:
   - "9092:9092"
  environment:
   KAFKA_BROKER_ID: 2
   KAFKA_LISTENERS: PLAINTEXT://broker2:9092
   KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker2:9092
   KAFKA_ADVERTISED_HOST_NAME: broker2
   KAFKA_ADVERTISED_PORT: 9092
   KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181
   #JMX_PORT: 9988
  volumes:
   - /var/run/docker.sock:/var/run/docker.sock
   - ./broker2:/kafka/kafka\-logs\-broker2
  external_links: # 连接本compose文件以外的container
  - zoo1
  - zoo2
  - zoo3
  networks:
   default:
    ipv4_address: 172.23.0.15

 broker3:
  image: wurstmeister/kafka
  restart: always
  hostname: broker3
  container_name: broker3
  privileged: true
  ports:
   - "9093:9092"
  environment:
   KAFKA_BROKER_ID: 3
   KAFKA_LISTENERS: PLAINTEXT://broker3:9092
   KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker3:9092
   KAFKA_ADVERTISED_HOST_NAME: broker3
   KAFKA_ADVERTISED_PORT: 9092
   KAFKA_ZOOKEEPER_CONNECT: zoo1:2181,zoo2:2181,zoo3:2181#
   #JMX_PORT: 9989
  volumes:
   - /var/run/docker.sock:/var/run/docker.sock
   - ./broker3:/kafka/kafka\-logs\-broker3
  external_links: # 连接本compose文件以外的container
  - zoo1
  - zoo2
  - zoo3
  networks:
   default:
    ipv4_address: 172.23.0.16

 kafka-manager:
  image: sheepkiller/kafka-manager:latest
  restart: always
  container_name: kafka-manager
  hostname: kafka-manager
  ports:
   - "9000:9000"
  links:      # 连接本compose文件创建的container
   - broker1
   - broker2
   - broker3
  external_links:  # 连接本compose文件以外的container
   - zoo1
   - zoo2
   - zoo3
  environment:
   ZK_HOSTS: zoo1:2181,zoo2:2181,zoo3:2181
   KAFKA_BROKERS: broker1:9092,broker2:9092,broker3:9092
   APPLICATION_SECRET: letmein
   KM_ARGS: -Djava.net.preferIPv4Stack=true
  networks:
   default:
    ipv4_address: 172.23.0.10

networks:
 default:
  external:  # 使用已创建的网络
   name: zoo_kafka
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

进入zookeeper客户端观察节点，校验是否启动成功

3 kafka中的基本概念

Broker:消息中间件的处理节点，一个Kafka就是一个broker节点，一个或则多个Broker可以组成，一个Kafka集群。
Topic:kafka根据Topic对消息进行归类，发布到Kafka集群中的消息必须指定一个Topic
Producer:消息生产者
Consumer:消息消费者
ConsumerGroup:每一个Consumer属于一个特定的Consumer Group，一条消息可以被不同的ConsumerGroup消费，但是一个ConsumerGroup中只能有一个Consumer能够消费该消息
Partition:物理上的概念，一个topic可以分为多个partition,每一个partition内部是有序的。

4 Kafka命令行

1. 创建主题
   kafka-topics.sh --create --zookeeper zoo3:2181 --replication-factor 1 --partitions 1 --topic test1
2. 查看主题
   kafka-topics.sh --list --zookeeper zoo3:2181
3. 发送消息(把消息发给broker中的某个Topic,打开kafka发送消息的客户端，然后开始用客户端向kafka服务器发送消息)
   kafka-console-producer.sh --broker-list broker1:9092 --topic test
   会打开一个客户端，每一行都是一个消息
4. 消费消息(打开一个消费消息的客户端，向kafka的某个主题消费消息)
   a、从当前主题中的最后一条消息的offset(偏移量位置)+1开始消费
   方式一：kafka-console-consumer.sh --bootstrap-server broker1:9092 --topic test 从尾部开始消费
   b、从当前主题的第一条开始消费
   方式二：kafka-console-consumer.sh --bootstrap-server broker1:9092 --from-beginning ----topic test 按偏移量消费
   关于消息的细节：
   1、生产者将消息发送给broker，broker会将消息保存在本地日志文件中。
   消息保存在Kafka节点下面。
   
   2、消息的保存是有序的，通过offsite来描述消息的有序性
   3、消费者消费消息也是通过offsite来消费消息

5 消息类型

5.1 单播消息

在一个Kafka的topic中，启动两个消费者，一个生产者，问：生产者生产消息，这条消息能够被两个消费者同时消费
**结论：如果消费者在同一个消费组，那么只有一个消费者可以收到订阅的topic的消息，换而言之，同一个消费组中只有一个消费者收到一个topic中的消息。**
以Group组的形式开启消费者客户端：
kafka-console-consumer.sh --bootstrap-server broker1:9092 --consumer-property group.id=testGroup --topic test
1
2
3
4

5.2 多播消息

kafka-console-consumer.sh --bootstrap-server broker1:9092 --consumer-property group.id=testGroup1 --topic test
不同的消费组订阅同一个topic,那么不同的消费组中只有一个消费者能收到消息，实际上也是多个消费组中的多个消费者收到同一个消息。
总结：一个topic类型的消息：在同一个消费组里只有一个消费者能消费，在不同消费组中每组都可以有一个消费者消费消息。

5.2 查看消费组及信息

1、查看一个Topic下面有哪些消费组
kafka-consumer-groups.sh --bootstrap-server broker1:9092 --list

2、查看消费组的详细信息
kafka-consumer-groups.sh --bootstrap-server broker1:9092 --describe --group testGroup

• current-offset:最后被消费的消息偏移量
• log-end-offset:消息总量
• lag:积压的消息数量

6 kafka主题和分区的概念

6.1 主题Topic

主题在Kafka中是一个逻辑的概念，Kafka通过Topic将消息进行分类，不同的Topic会被订阅不同Topic的消费者消费
:Topic就相当与把消息贴上不同的标签进行分类，好提供给下游订阅该分类的者消费者消费。

6.2 分区partition

但是有一个问题，如果说这个topic中的消息非常多，TB级别单位，因为消息会被保存在log日志中，这时候为了解决文件过大的问题，Kafka提出了分区的概念
通过partition将一个topic中的消息进行分区存储，这样好处：
1、分区存储，可以解决统一存储文件过大的问题
2、提高了读写的吞吐量，读写可以在不同的分区中进行。
Kafka内部自己创建了_consumer_offset分区，用来保存消费者消费的偏移量。
为什么设置50个分区？并行操作，提高性能。
因为每个消费者自己维护着消费的主题的偏移量，那也就是每个消费者会把消费的主题自主提交给Kafka
因此为了提高并发量，Kafka默认设置50个分区，提高并发量。
提高写与消费的效率
1、创建一个分区test2有两个分区
kafka-topics.sh --create --zookeeper zoo3:2181 --replication-factor 1 –partitions 2 --topic test2

6.3 副本的概念

副本是为主题中的分区创建的备份，多个副本在Kafka中的不同集群上的多个broker上，会有一个副本作为leader其他的是follower
副本就是对于分区的备份，在集群中不同的副本会部署在不同的broker上。
1、kafka-topics.sh --create --zookeeper zoo3:2181 –replication-factor 3 –partitions 2 --topic test2
查看topic详细信息
2、kafka-topics.sh --describe --zookeeper zoo1:2181 --topic test4

Topic主题下面有两个分区，每个分区有三个副本。副本保证了高可用。
每个分区都有一个Broker作为Leader,
leader：kafka的写和读都发生在leader上，leader负责把数据同步给follower，当leader挂了，经过主从选举，会从多个follower选举出一个leader
follower：接收leader的同步数据
isr:可以同步，和以及同步的节点，存放在isr集合中，如果isr中的节点性能比较差，Kafka会把它从这个集合中删除。
集群中有多个broker，创建主题时候可以指明主题有多个分区(把消息拆分到不同的分区存储)
可以为分区创建不同的副本，不同副本可以放在不同的broker中存储。

6.3 分区消费的细节

1、一个topic分区只能被某一个消费者组中的一个消费者消费。(目的是为了保证消费的顺序性，)
2、partition决定了消费组中消费者的数量，建议消费组中的消费者不要超过partition数量
3、如果消费者挂了，那么会触发rebalance机制，会让其他消费者来消费该分区

7 Kafka的API操作

7.1 同步和异步发送消息

1、同步发送，生产者向Kafka消息队列发送消息，如果kafka没有ack,生产者会进行重试
重试3次，每次6秒。
2、异步发送，Callback()回调方法，主线程发完消息之后，回调方法去接收消息是否接收成功。
异步发送：生产者发送完消息之后就可以执行之后的业务，Callback负责接收broker消息是否接收成功。

7.2 关于Ack消息参数配置

ack = 0：不用确定是否收到 立即返回，容易都是消息，效率高
ack = 1:多个副本之间leader已经收到消息，并把消息写到本地log中，才会返回生产者。性能最均
ack = -1/all :里面有默认的配置min.replicas=2（默认为1，推荐大于等于2）需要有一个leader与follower同步完成之才返回安全性能最高性能差。

7.2 消息缓冲区

kafka默认会创建一个消息缓冲区，用来存放需要发哦是那个的消息
生产者会有一个缓冲区：默认32M
pros.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
Kafka线程从缓冲区内区拿数据，一次拿16K，然后放进kafka队列。
pros.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
问题：如果数据没有达到16k,10毫秒以后将数据发送。
pros.put(ProducerConfig.LINGER_MS_CONFIG,10);

7.3 offset自动提交与手动提交

自动提交开关。
pros.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,“false”);
提交内容:消费者无论是手动提交还是自动提交，都需要把所属的消费组+消费的某个主题以及消费的分区，消费的偏移量这样的消息提交到集群中_consumer_offsets主题中。
自动提交：1、消费者消费消息的时候是主动拉取消息，
2、消息拉取下来的时候直接提交offset
自动提交有可能消息会丢失，丢失的时间是一个消费者拉取消息之后手动提交之后，立马挂掉之后
手动提交：1、消费消息时或则后，手动提交offset。
手动同步提交：在消费完消息后调用同步提交的方法，当集群返回ack之前一致阻塞，收到ack之表示提交成功，执行之后的逻辑。一般使用同步提交。
手动异步提交： 在消息消费完之后，不需要等到集群回应，可以执行之后的逻辑，可以设置一个回调方法供集群调用。

7.4 长轮询poll消息

1、默认情况下，消费者一次会poll500条消息
2、代码中设置长轮询的时间是1000毫秒
3、如果一次poll到500条，直接执行消费。
4、如果一次没有poll到500条，且时间在1秒内，那么长轮询会继续poll,要么到500条，要么到1秒
5、如果多次poll都没有达到500条，且一秒的时间到了，那么执行消费
就是说，一次poll500条消息，如果一次poll到就开始消费，如果没有到500条就继续poll时间是1秒。

如果消费者两次poll的时间超过30秒，则集群认为该消费者消费能力，则就会提出消费者组。

7.5 消费者的健康状态检查

pros.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG,1000);
消费者每隔1秒向Kafka集群发送心跳，集群如果发现有超过10秒没有续约的消费者，将会被踢出消费者组，触发消费组的rebalance机制，将该分区交给消费者真的其他消费者进行消费。

7.6 消费方式

1、指定分区，
2、从头消费
2、指定offset消费
3、指定时间点消费。找到时间节点的偏移量，然后按偏移量消费。

8 SpringBoot中整合使用Kafka

8.1 多个主题指定消费

9 Kafka集群中Controller，rebalance,HW作用

9.1 controller

1、集群中的controller由谁来充当
每个broker启动时会向zk创建一个临时的序号节点，获得序号最小的broker将会作为集群中的controller
controller：负责选举leader,通知其他节点更新ISR信息
1、负责管理集群中的分区，当集群中的leader挂掉之后负责选取新的leader副本。选取机制：在Isr列表最左边的broker作为leader。
2、当检测到某个分区的ISR集合发生变化的时候，通知其他broker更新元信息。

9.2 rebalance

触发rebalance的前提：消费者没有明确指定分区消费，当消费组里的消费者关系发生变化的时候，那么就会触发rebalance机制。这个机制就是重新调整消费者消费哪个分区
1、range:通过公式计算消费者消费哪个分区
2、通过轮询
3、sticky：在触发rebalance之后，在消费者消费原分区不变的基础上进行调整。
如果没有开启sticky策略，所有连接会重新断掉再分配，如开开启了再原来关系不变的基础上进行分配。

9.3 HW和LEO

HW:消费者只能消费同步完成的数据。解决问题：防止数据丢失。
LEO:是某个副本最后消息的位置(log-end-offset)

10 kafka线上问题优化

10.1 如何防止消息丢失？

发送方：ack设置为1或则-1/all可以防止消息丢失，并且设置同步的分区大于2
消费方：可以把自动提交修改为手动提交

10.2 如何防止消息的重复消费？

场景一：由于网络不稳定造成消息重新发送，导致消费方消息重复消费。
生产者：关闭重复发送 --会导致丢消息
消费者：幂等性的保证 对于rest的请求(get(),post(非幂等),put(),delete())
1、在MySQL中创建联合主键，联合主键(id主键自增，业务id)防止多条主键创建相同记录，保证不被重复消费
2、redis的分布式锁，保证只有一条记录创建成功。

10.3 Kafka如何做到顺序消费？

场景：订单的顺序消费，订单创建，扣减库存，支付等
生产者创建消息：123，消费者消费顺序123
生产方：同步发送，ack不能设置为0，关闭重试。保证发送顺序一致
消费者：方案：使用一个partition使用一个消费者
缺点：牺牲性能
使用场景不多，Rockermq封装了这一块的功能。

10.3 Kafka如何解决消息挤压问题？

堆积：消息在Kafka集群中没有被消费。
消息积压后消费者寻址花费的时间就很长。
1、消息积压问题出现
消息的消费者的消费速度赶不上生产者的生产速度。导致Kafka中大量的数据没有被消费，随着没有被消费的消息积压越多，消费者寻址性能就会变得很差，导致Kafka对外提供的服务性能很差，重而导致其他的服务访问速度变慢。造成服务雪崩。
解决方案：
1）在这个消费者中使用多线程，充分利用机器的性能进行消费消息
2} 创建多个消费组，多个消费者，部署到其他机器上，提高消费速率。
3）创建一个消费者，该消费者在Kafka上新建一个主题创建多个partition分区，然后多个消费者进行消费
4）通过业务架构的设计，提升消费层面的消费能力。

10.4 延迟队列

使用场景：在订单创建超过30分钟没有付款就取消订单，这种效果可以使用延迟队列来实现
方案一：kafka中创建相应的主题，消费者消费该主题的主题，消费者消费消息时候会判断创建时间是否超过30分钟
是：去数据库中修改订单的状态
否：记录当前消息的offset，并不在继续消费之后的消息，等待一分钟，再次向Kafka拉取offset之后的消息