Kafka-为什么用消息队列、消息队列解决的实际问题、MQ的几种类型、kafka的基本概念及单播多播使用、kafka的topic和partition、2个面试题

为什么使用消息队列

试想一下，假如我们有一个电商系统。用户下了一个订单，然后系统需要做什么？

新增一条订单 (订单服务)
商品库存 -1 或其他数量 (库存服务)
用户积分 (积分服务)
优惠券 (优惠券服务)
…

同步方式

执行了一步下订单，然后后面有很多的操作需要系统去执行，简单点执行就是我们可以将后面这些操作全都放入到一个方法中执行，同步执行它们，执行完才能说这个订单创建完毕了。

同步执行存在的问题：

造成系统的开销-响应时间太长
网络突然出现异常时，订单创建会失败

同步执行的方式会存在性能和稳定性的问题。

异步方式

我们可以把全部在一个方法中执行的操作，给存放到一个中间件中，也就是消息队列。用来存放一些消息，当某个服务监听到这个消息的时候，这个服务就去执行相应的操作。比如库存服务接收到一条减库存的消息，然后库存服务就去执行了。订单服务在把订单创建好以后，直接给相应的服务发消息，其他服务收到消息后执行相应的操作。

其他服务执行就是异步执行的，不会造成响应时间太长的问题。

异步执行的优势：

明显的提升系统的吞吐量
即使有服务失败，也可以通过分布式事务解决方案保证最终的数据一致性

针对于同步的通信方式来说，异步的方式，可以让上游快速成功，极大提高了系统的吞吐量。而且在分布式系统中，通过下游的分布式事务也能保障业务执行之后的最终一致性。

消息队列解决的问题就是----通信问题。无论是dubbo的rpc调用，还是cloud的restTemplate调用都是同步的通信方式，消息队列就是异步的通信。

消息队列的类型什么是MQ？

Message Queue(MQ)，消息队列中间件。首先MQ给人的直接感觉就是实现程序的异步和解耦，但是MQ真正的目的是为了通信，屏蔽底层复杂的通讯协议，定义更加简单的通信协议。

一个分布式系统中多个模块之间的通信要么是HTTP，要么就是(rpc)TCP，但是这两种协议体积过于庞大，必须要携带Cookie、状态码、响应码…消息中间件自己定义了一套协议更加轻量级。

消息中间件的选型

目前消息队列的中间件选型有很多：

RabbitMQ ：功能性强
RocketMQ ：根据kafka内部执行原理，模仿的一个消息队列中间件。性能与kafka比肩。功能比kafka多。
Kafka ：消息处理性能最快的一款MQ
ZeroMQ

他们有什么区别？

总体上可以分为两大类：有Broker和无Broker

Broker是什么呢？Broker就是指一个中间件的服务器，很多大型的项目中，都会用到某些中间件的集群，为了保证服务的高可靠。多个Broker组合在一起就成为了一个中间件集群。

有Broker

这中类型的中间件通常有一台服务器作为Broker，所有的消息都通过它中转。生产者把消息发送给它就结束自己的任务了，Broker则把消息主动推送给消费者(或者消费者主动轮询)

重Topic

Kafka、ActiveMQ就属于重Topic，生产者发送key和数据到Broker，由Broker比较key之后决定给哪个消费者。这种是比较常见的模式，此模式下一个Topic往往是一个比较大的概念，用于消息分类。甚至一个系统中可能只有一个Topic，Topic某种意义上就是queue。

整个Broker，根据topic进行消息的中转。在重Topic的消息队列中必然需要topic存在

轻Topic

如RabbitMQ，它的topic只是中转的一种模式，它还有其他的模式比如routingKey等。轻Topic的中间件，topic可以不存在。

无Broker

无Broker的MQ的一个代表ZeroMQ。其作者敏锐认识到–MQ就是更高级的Socket，它是用来解决通信问题的。所以ZeroMQ被设计为一个库，而不是中间件。中间没有服务器作为中转，而是生产者和消费者直接进行通信。

节点之间通信的消息都是发送到彼此的队列中，每个节点既是生产者又是消费者。ZeroMQ做的事就是封装了一套类似于Socket的API可以完成发送数据，读取数据。

kafka

kafka是最初由Linkedin公司开发，是一个分布式、支持分区(partition)、多副本(replication)，基于Zookeeper协调的分布式消息系统。

使用场景：

日志收集：kafka很适合用作日志收集，通过kafka以统一接口服务的方式开放给各种consumer。
消息系统：解耦和消息缓存，生产者消费者
用户活动跟踪：kafka常用来记录web用户或app用户各种活动，如浏览网页、搜索、点击等活动。这些信息被各个服务发布到kafka的topic中，订阅者通过订阅这些topic来做实时监控分析。

下载安装启动

下载kafka，kafka下载地址

下载完解压缩后，会得到一个文件夹，内容如下：以windows系统为例

bin目录下都是一些可执行的目录，由于kafka依赖到Zookeeper，这个目录中Zookeeper相关的一些启动文件也都在这里
config目录中都是一些配置文件。包含Kafka和Zookeeper的配置

看一下kafka的配置文件，里面大部分都是已经配置好了，我们可以根据需要做一些更改：

给大家看一些比较重要的配置

############################# Server Basics #############################
# kafka集群中，用来区分某一个 broker节点的 标识，不能重复
broker.id=0

############################# Socket Server Settings #############################
# 这里配置kafka的连接地址，不做配置的话就是本机localhost(可以不做任何配置，注释掉就行)
listeners=PLAINTEXT://localhost:9092

############################# Log Basics #############################
# 存放kafka消息数据的地址
log.dirs=/tmp/kafka-logs

# 分区数量
num.partitions=1


############################# Zookeeper #############################
# 配置Zookeeper的连接地址
zookeeper.connect=localhost:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

确保kafka所在的机器上安装了jdk，否则会出问题

启动Zookeeper

binwindowszookeeper-server-start.bat configzookeeper.properties

一定要在启动完Zookeeper之后再去启动kafka，因为kafka的服务需要注册到Zookeeper上面进行管理

启动kafka

binwindowskafka-server-start.bat configserver.properties

上述为windows启动，linux系统启动的话到bin目录下找即可。不用再加 windows的目录了。

基本概念

名称	解释
Broker	消息中间件的处理节点，一个Kafka节点就是一个Broker，一个或多个Broker可以组成一个kafka集群
Topic	kafka根据Topic进行消息分类，发布到kafka集群的每条消息需要指定一个Topic
Producer	消息生产者，向Broker发送消息的客户端
Consumer	消息消费者，向Broker读取消息的客户端
Consumer Group	每个Consumer属于一个特定的Consumer Group，一条消息可以被多个不同的Consumer Group消费，但是一个Consumer Group中只有一个Consumer能消费这个消息
Partition	物理上的概念，一个Topic可以分为多个Partition，每个Partition内部消息有序

基本使用

注意：以下都是在 windows环境中的命令，如果使用的是linux系统，把执行文件的后缀改下就行。

创建主题，创建一个名为 test的 topic主题，这个topic只有一个 partition，且备份也为1(windows中为.bat文件，linux中是sh文件)

.kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

查看当前kafka内有哪些topic

.kafka-topics.bat --list --zookeeper localhost:2181

发送消息到test的topic中

创建生产者生产消息

创建一个生产者，–broker-list 是指 kafka服务器的地址(可以是多个服务器地址)

.kafka-console-producer.bat --broker-list localhost:9092 --topic test

创建完之后，我们可以在里面发一些消息

这些消息会被存入到kafka的服务器中去。

创建消费者进行消息

创建消费者，指定消费者监听哪个kafka服务，哪个topic主题下的消息

消费者客户端有两种方式进行消费：

第一种：从最后一条消息的偏移量(offset) + 1 开始消费
```
.kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test
```
这种方式的意思是：从现在开始进行监听，只监听最新的消息，在监听前kafka中如果已经有了消息，那么消费者不会去进行消费，只有当生产者再次生产消息时，消费者才会进行消费

第二种：从头开始消费
```
.kafka-console-consumer.bat --bootstrap-server localhost:9092 --from-beginning --topic test
```
此方式是把此topic中的所有消息全部从头进行消费

需要注意几个点：

消息会被存储，存储到kafka的log.dirs配置的目录中
消息是有序的，通过offset描述消息的有序性
消费者消费消息也是通过offset来找到当前需要消费的消息的位置

消费模式单播、多播

假如有多个消费者，消费同一个topic中的消息，两个消费者能否同时消费到这个消息？

消息单播

一个消费者组中只有一个消费者能够消费到某一个topic中的消息。可创建多个消费者在同一个组中

.kafka-console-consumer.bat --bootstrap-server localhost:9092 --consumer-property group.id=testGroup1  --topic test

以上命令创建的多个消费者，只有一个消费者能够消费到消息。其实，更严谨点说，应该是一个消费者组中只能有一个消费者消费同一个 topic下的同一个partition，而默认创建的topic中，只有一个partition，所以这里说，一个消费者组只有一个消费者能消费到同一个topic中的数据。

消息多播

消息多播，如果要让所有的消费者都能消费到消息，那么把他们放置到不同的消费者组中即可。

.kafka-console-consumer.bat --bootstrap-server localhost:9092 --consumer-property group.id=testGroup1  --topic test

.kafka-console-consumer.bat --bootstrap-server localhost:9092 --consumer-property group.id=testGroup2  --topic test

查看消费者组及信息

# 查看消费组列表
.kafka-consumer-groups.bat --bootstrap-server localhost:9092 --list

# 查看某个消费组的具体信息，当前偏移量、最后一条消息偏移量、堆积消息数量
.kafka-consumer-groups.bat --bootstrap-server localhost:9092 --describe --group testGroup1

Partition : 分区号
current-offset ：当前偏移量
log-end-offset : 最新偏移量
lag : 剩余几个偏移量的消息没被消费

主题和分区主题 topic

主题topic在kafka中是一个逻辑的概念，kafka通过topic进行消息分类。不同的topic会被订阅该topic的消费者消费。但是，如果topic中有很多的消息，多到几个T，因为消息是被保存到log的日志文件中的，为了解决文件过大问题，kafka提出了partition分区的概念。

分区 partition

一个topic中消息量是很大的，可以通过分区来设置，分布式进行存储这些消息。分段式存贮。比如一个topic中有3个partition，那么发送的消息会被分别存入到这三个partition中的。

而且，写入到不同的partition时，是可以并行执行的，所以大大提高了效率，提高了服务系统的吞吐量(tps)。

分区存储，解决统一存储文件过大问题
提供了读写的吞吐量：读和写可以同时在多个分区中进行

创建一个多分区的topic。

.kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 2 --topic test1

然后在kafka的日志数据目录中，就会发现多了两个文件夹 test1-0 和 test1-1，这两个目录就是topic为 test1的两个分区目录，每个目录中都有.log .index .timeindex的文件，理解为分段式(segement)。

,index就是索引文件，可以快速查找的

kafka日志数据目录

kafka日志数据目录下有很多的类似 _consumer_offsets-0这种文件夹，是用来干嘛的呢？

kafka内部自己创建了50个 _consumer_offsets主题的分区，这个主题是用来存放消费者消费到的偏移量位置。

那么这个偏移量是怎么进行维护呢？

消费者消费完消息后，需要给kafka服务器的_consumer_offsets主提中提交一个offset的偏移量，然后kafka本地中将这个偏移量记录起来

消费者如何知道自己应该从哪里消费呢？

消费者每次消费时，需要先到_consumer_offsets主题中获取自己上次消费到的偏移量，然后继续进行消费

细节点：

定期将自己消费的offset提交给kafka内部的topic: _consumer_offsets。提交过去的时候，key是consumerGroupId+topic+分区号码，value是当前的offset值。
因为_consumer_offsets可能会接收高并发的请求，kafka默认分配给它50个分区，提高吞吐量
通过公式选出consumer消费的offset要提交到哪个 consumer_offsets分区中：
```
hash(consumerGroupId) % _consumer_offsets主题的分区数量
```
文件中保存的消息，默认保存7天，7天后自动删除。(可去配置文件中进行配置)

面试题

面试题:消息中间件为什么不使用http协议？

http协议实际上是在tcp/ip协议基础上构建的，请求响应报文中包含一些cookie数据，状态码，响应码等过于复杂了，对我们的消息而言，根本用不到，我们只需要保证消息传递快捷迅速即可(传递、存储、分发)。
http协议是一个短链接模式，如果我们的连接中断了，那么数据都会丢失，无法持久化，而我们的消息需要考虑这种持久化保证消息的高可靠、可用。

面试题：常见的一些消息中间件协议：

AMQP：高级消息队列协议，特性：

支持分布式事务
支持消息持久化
高可用，高性能消息处理优势
支持：RabbitMQ、ActiveMQ中间件

MQTT：即时通讯协议，是物联网系统架构的重要组成

不支持事务
不支持持久化
轻量、简单
适用于：低宽带，网络不稳定
支持：RabbitMQ(需要打开协议支持)、ActiveMQ中间件

OpenMessage：阿里、滴滴、雅虎等创建的一套协议

支持事务
支持持久化
简单，解析速度块
支持：RocketMQ中间件

Kafka：kafka协议(apache)，基于TCP/IP的二进制传输协议

速度最快
支持持久化
kafka从0.11版本加入了支持事务
支持：Kafka中间件

化，而我们的消息需要考虑这种持久化保证消息的高可靠、可用。`