一、为什么使用消息队列

1.使用同步的通信方式来解决多个服务之间的通信

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

2.使用异步的通信方式

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

消息队列解决具体的是什么问题——通信问题。

二、消息队列的流派

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

• rabbitMQ：内部的可玩性（功能性）是非常强的
• rocketMQ：阿里内部一个大神，根据kafka的内部执行原理，手写的一个消息队列中间件。性能是与kafka比肩，除此之外，在功能上封装了更多的功能。
• kafka：全球消息处理性能最快的一款MQ
• zeroMQ

这些消息队列中间件有什么区别？

1.有broker

• 重topic：kafka、RocketMQ、ActiveMQ

  整个broker，依据topic来进行消息的中转。在重topic的消息队列里必然需要topic的存在。

• 轻topic：RabbitMQ

  topic只是一种中转模式。

2.无broker

在生产者和消费者之间没有使用broker，例如zeroMQ，直接使用socket进行通信。

三、Kafka的基本知识

1.Kafk2a的安装

• 部署一台zookeeper服务器

• 安装jdk

• 下载kafka的安装包：https://kafka.apache.org/downloads

• 上传kafka到服务器上：/user/local/kafka

• 解压缩压缩包

• 进入到config目录内，修改server.properties

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

• 进入到bin目录内，执行以下命令来启动kafka服务器（带着配置文件）

  ./kafka-server-start.sh -daemon ../config/server.properties
  

• 校验kafka是否启动成功：

  进入到zk内查看是否有kafka的节点：brokers/ids/0

2.Kafka中的一些基本概念

kafka中有这么些复杂的概念

名称	解释
Broker	消息中间件处理节点，一个Kafka节点就是一个broker，一个或多个Broker可以组成一个Kafka集群
Topic	Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic
Producer	消息生产者：向Broker发送消息的客户端
Consumer	消息消费者：从Broker读取消息的客户端

3.创建topic

• 通过kafka命令向zk中创建一个主题

  ./kafka-topics.sh --create --zookeeper ip:2181 --replication-factor 1 --partitions 1 --topic test
  

• 查看当前zk中所有的主题

  ./kafka-topics.sh --list --zookeeper ip:2181
  test
  

4.发送消息

把消息发送给broker中的某个topic，打开一个kafka发送消息的客户端，然后开始用客户端向kafka服务器发送消息

./kafka-console-producer.sh --broker-list ip:9092 --topic test

5.消费消息

打开一个消费消息的客户端，向kafka服务器的某个主题消费信息

• 方式一：从当前主题中的最后一条消息的offset（偏移量位置）+1开始消费

  ./kafka-console-consumer.sh --bootstrap-server ip:9092 --topic test
  

• 方式二：从当前主题中的第一条消息开始消费

  ./kafka-console-consumer.sh --bootstrap-server ip:9092 --from-beginning --topic test
  

6.关于消息的细节

• 生产者将消息发送给broker，broker会将消息保存在本地的日志文件中

  /usr/local/kafka/data/kafka-logs/主题-分区/000000000.log
  

• 消息的保存是有效的，通过offset偏移量来描述消息的有序性

• 消费者消费信息时也是通过offset来描述当前要消费的那条消息的位置

7.单播消息

在一个kafka的topic中，启动两个消费者，一个生产者。问：生产者发送消息，这条消息是否同时会被两个消费者消费？

如果多个消费者在同一个消费组，那么只有一个消费者可以收到订阅的topic中的消息。换言之，同一个消费组中只能有一个消费者收到一个topic中的消息。

./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup --topic test

8.多播消息

不同的消费组订阅同一个topic，那么不同的消费组中只有一个消费者能收到消息。实际上也是多个消费组中的多个消费者收到了同一个消息。

#消费组testGroup1
./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup1 --topic test
#消费组testGroup2
./kafka-console-consumer.sh --bootstrap-server ip:9092 --consumer-property gropu.id=testGroup2 --topic test

9.查看消费组的详细信息

通过以下命令可以查看到消费组的详细信息：

./kafka-consumer-groups.sh --bootstrap-server ip:9092 --describe --group testGroup

重点关注以下几个信息：

• Currennt-offset：当前消费组的已消费偏移量
• Log-end-offset：主题对应分区消息的结束偏移量（HW）
• Lag：当前消费组未消费的消息数

四、Kafka中主题和分区的概念

1.主题Topic

主题-topic在kafka中是一个逻辑的概念，kafka通过topic将消息进行分类。不同的topic会被订阅该topic的消费者消费。

但是有一个问题，如果说这个topic中的消息非常非常多，多到需要几T来存，因为消息是会保存到log日志文件中的。为了解决这个文件过大的问题，kafka提出了Partition分区的概念。

2.分区Partition

1）分区的概念

通过partition将一个topic中的消息来存储。这样的好处有多个：

• 分区存储，可以解决统一存储文件过大的问题。

• 提供了读写的吞吐量，读和写可以同时在多个分区中进行。

  

2）创建多分区的主题

./kafka-topics.sh --create --zookeeper ip:2181 --replication-factor 1 --partitions 2 --topic test1

3.kafka中消息日志文件中保存的内容

• 00000.log：这个文件中保存的就是消息。

• _consumer_offsets-49

  kafka内部自己创建了_consumer_offset主题包含了50个分区。这个主题用来存放消费者消费某个主题的偏移量，因为每个消费者都会自己维护消费的主题的偏移量，也就是说每个消费者会把消费的主题的偏移量自主上报给kafka中的默认主题：consumer_offsets。因为kafka为了提升这个主题的并发性，默认设置了50个分区。

  • 提交到哪个分区：通过hash函数确定：hash（consumerGroupId）%_consumer_offsets主题的分区数。
  • 提交到该主题中的内容是：key是consumerGroupId+topic+分区号，value就说当前offset的值。

• 文件中保存的消息，默认保存7天，7天到后消息会被删除。

五、Kafka集群操作

1.搭建kafka集群（三个broker）

• 创建三个server.properties文件

  # 0 1 2
  broker.id=2
  # 9092 9093 9094
  listeners=PLAINTEXT://192.168.65.60:9004
  # kafka-logs kafka-logs-1 kafka-logs-2
  log.dir=/usr/local/data/kafka-logs-2
  

• 通过命令来启动三台broker

  ./kafka-server-start.sh -daemon ../config/server.properties
  ./kafka-server-start.sh -daemon ../config/server1.properties
  ./kafka-server-start.sh -daemon ../config/server2.properties
  

• 校验是否启动成功

  进入到zk中查看/brokers/ids中是否有3个znode（0，1，2）

2.副本的概念

在创建主题时，除了指明主题的分区数外，还指明了副本数，那么副本是一个什么概念呢？

副本是为了给主题中的分区创建多个备份，多个副本在kafka集群的多个broker中，会有一个副本作为leader，其他是follower。

• leader：kafka的写和读的操作都发生在leader上。leader负责把数据同步给follower。当leader挂了，经过主从 选举，从多个follower中选举产生一个新的leader。
• follower：接受leader同步的数据。
• isr：可以同步和已同步的节点会被存入到isr集合中。细节：如果isr中的节点性能比较差，会被踢出isr集合。

（重点~！）此时，broker、主题、分区、副本这些概念就全部展现了，大家需要把这些概念梳理清除：

集群中有多个broker，创建主题时可以指明主题有多个分区（把消息拆分到不同的分区中存储），可以为分区创建多个副本，不同的副本存放在不同的broker里。

3.关于集群消费

1）向集群发送消息

./kafka-console-producer.sh --broker-list ip1:9092,ip1:9093,ip1:9094 --topic my-replicated-topic

2）从集群消费消息

./kafka-console-consumer.sh --bootstrap-server ip1:9092,ip1:9093,ip1:9094 --from-beginning --topic my-replicated-topic

3）指定消费组来消费消息

./kafka-console-consumer.sh --bootstrap-server ip1:9092,ip1:9093,ip1:9094 --from-beginning --consumer-property group.id=testGroup1 --topic my-replicated-topic

4）分区分消费组的集群消费中的细节

• 一个partition只能被一个消费组中的一个消费者消费，目的是为了保证消费的顺序性，但是多个partition的多个消费者消费的总的顺序性是得不到保证的，那怎么做到消费的总顺序性呢？
• partition的数量决定了消费组中消费者的数量，建议同一个消费组中消费者的数量不要超过partition的数量，否则多个消费者没有消息消费。
• 如果消费者挂了，那么会触发rebalance机制，会让其他消费者来消费该分区。