一、消息队列Message Queue

image.png

两种模式

1. 点对点模式
   点对点模式是一个基于拉取或轮询的消息传送模型，由消费者主动拉取数据，客户端需要实时开启一个线程监控队列中是否有数据。
2. 发布/订阅模式
   发布/订阅模式是一个基于推送的消息传送模型，由MQ主动推送消息给所有订阅者，即使当前订阅者不可用。

优点

• 解耦
  允许你独立的扩展或修改两边的处理过程，只要确保它们遵循同样的接口约束。
• 冗余
  MQ把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。
• 扩展
  因为MQ解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可，提高了灵活性和峰值处理能力。
• 可恢复性
  消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。
• 顺序保证
  队列本身就遵循FIFO的原则，保证了数据的处理顺序。
• 缓冲
  有助于控制和优化数据流经过系统的速度，解决生产者和消费者处理速度不一致的问题。
• 异步通信
  MQ提供了异步处理机制，允许用户把消息放入队列但不立即处理它，直到需要时再去处理。

二、Kafka简介

1. Apache Kafka是一个开源消息系统，由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。
2. Kafka最初是由LinkedIn公司开发，并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。
3. Kafka是一个Distributed Message Queue。Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接收者称为Consumer，此外Kafka集群有多个Kafka实例组成，每个实例称为Broker。
4. Kafka需要依赖于Zookeeper保存一些Meta信息，来保证系统可用性。
5. 在流式计算中，Kafka一般用来缓存数据，Storm通过消费Kafka的数据进行计算。

三、Kafka架构

image.png

No.	组件	说明
1	Producer	消息生产者，向Kafka中发消息的客户端。
2	Consumer	消息消费者，从Kafka中取消息的客户端。
3	Broker	一台Kafka服务器就是一个Broker，一个集群由多个Broker组成，一个Broker可以容纳多个Topic。
4	Topic	属于特定类别的消息流称为Topic， 数据存储在Topic中，可以理解为一个队列。
5	Partition	为了实现负载均衡和高并发，一个非常大的Topic可以通过分为多个Partition分布到多个Broker上，每个Partition都是一个有序的队列。
6	Consumer Group（CG）	这是Kafka用来实现一个消息的广播（发给所有Consumer）和单播（发给任意一个Consumer）的手段。一个Topic可以有多个CG，Topic中的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个Partition只会把消息发给该CG中的一个Consumer。如果需要实现广播，只要每个Consumer都有一个独立的CG即可，需要实现单播，只要所有Consumer都在同一个CG即可。用CG还可以将Consumer进行自由分组而不需要重复发送消息到不同的Topic。
7	Offset	Partition中的每条消息都会被分配一个有序的id（Offset）。Kafka只保证每个Partition中的顺序，不保证多个Partition的顺序。
8	Leader	负责给定Partition的所有读取和写入的节点。
9	Follower	跟随Leader指令的节点被称为Follower。 如果Leader节点宕机，其中一个Follower将通过选举自动成为新的Leader。

四、Kafka工作流程

Producer生产过程

1. 写入方式
   Producer采用Push模式将消息发布到Broker，每条消息都被追加（append）到Partition中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障Kafka吞吐率）。

2. 分区（Partition）
   消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成，其组织结构如下图所示：

   
   image.png
   image.png
   

   我们可以看到，每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值。

   1. 分区的原因
      1. 负载均衡，方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个Topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了。
      2. 可以提高并发，因为可以以Partition为单位读写了。
   2. 分区的原则
      1. 已指定Partition，则直接使用该Partition。
      2. 未指定Partition但指定了Key，则通过对Key进行哈希计算得出一个Partition。
      3. Partition和Key都未指定，则轮询选出一个Partition。
3. 副本（Replication）
   同一个Partition可能会有多个Replication（对应server.properties配置中的default.replication.factor=N）。没有Replication的情况下，一旦Broker宕机，其上所有Partition的数据都不可被Consumer消费，同时Producer也不能再将数据存于其上的Partition。引入Replication之后，同一个Partition可能会有多个Replication，而这时需要在这些Replication之间选出一个Leader，Producer和Consumer只与这个Leader交互，其它Replication作为Follower从Leader中复制数据。

4. 写入流程

   
   image.png

Broker保存过程

1. 存储方式
   物理上把Topic分成一个或多个Partition（对应server.properties配置中的num.partitions=3），每个Partition物理上对应一个文件夹（该文件夹存储该Partition的所有消息和索引文件），如下：

[atguigu@hadoop102 logs]$ ll
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:37 first-0
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:35 first-1
drwxrwxr-x. 2 atguigu atguigu  4096 8月   6 14:37 first-2
[atguigu@hadoop102 logs]$ cd first-0
[atguigu@hadoop102 first-0]$ ll
-rw-rw-r–. 1 atguigu atguigu 10485760 8月   6 14:33 00000000000000000000.index
-rw-rw-r–. 1 atguigu atguigu      219 8月   6 15:07 00000000000000000000.log
-rw-rw-r–. 1 atguigu atguigu 10485756 8月   6 14:33 00000000000000000000.timeindex
-rw-rw-r–. 1 atguigu atguigu        8 8月   6 14:37 leader-epoch-checkpoint

2. 存储策略
   无论消息是否被消费，Kafka都会保留所有消息。有两种策略可以删除旧数据：

   1. 基于时间：log.retention.hours=168
   2. 基于大小：log.retention.bytes=1073741824

   需要注意的是，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高Kafka性能无关。

3. Zookeeper存储结构

   
   image.png
   

   注意：只有Broker和Consumer在Zookeeper中注册，Producer不在Zookeeper中注册。

Consumer消费过程

1. 消费者组

   
   image.png
   

   Consumer是以Consumer Group的方式工作，由一个或者多个Consumer组成一个Group，共同消费一个Topic。每个Partition在同一时间只能由Group中的一个Consumer读取，但是多个Group可以同时消费同一个Partition。在图中，有一个由三个Consumer组成的Group，有一个Consumer读取Topic中的两个Partition，另外两个Consumer分别读取一个Partition。某个Consumer读取某个Partition，也可以叫做某个Consumer是某个Partition的拥有者。
   在这种情况下，Consumer可以通过水平扩展的方式同时读取大量的消息。另外，如果一个Consumer失败了，那么Group中其它的Consumer会自动负载均衡读取之前失败的Consumer读取的Partition。

2. 消费方式
   Consumer采用Pull模式从Broker中读取数据。
   Push模式很难适应消费速率不同的Consumer，因为消息发送速率是由Broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成Consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而Pull模式则可以根据Consumer的消费能力以适当的速率消费消息。
   对于Kafka而言，Pull模式更合适，它可简化Broker的设计，Consumer可自主控制消费消息的速率，同时Consumer可以自己控制消费方式——即可以批量消费也可以逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。
   Pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，可以Pull请求中设置参数，允许Consumer请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）。

五、Kafka案例

生产者

package com.fas.kafka.service;

import org.apache.log4j.Logger;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import com.fas.kafka.util.KafkaUtil;

@Service
@Log4j
public class KafkaProducerService {
    private KafkaTemplate<String, String> kafkaTemplate = KafkaUtil.getKafkaTemplate();

    public void send(String topic, String data) {
        ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topic, data);
        listenableFuture.addCallback(result -> log.info("Send data to kafka [success] - " + topic + " - " + data), result -> log.info("Send data to kafka [error] - " + topic + " - " + data));
    }
}

消费者

package com.fas.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.MessageListener;

import com.alibaba.fastjson.JSONObject;
import com.fas.kafka.entity.DeviceOnlineData;

import lombok.extern.log4j.Log4j;

@Log4j
public class KafkaConsumer implements MessageListener<String, String> {
    public static final BlockingQueue<DeviceOnlineData> DATA_ACCESS_BLOCKING_QUEUE = new LinkedBlockingQueue<>();
    public static final String KAFKA_DATA_ACCESS_TOPIC = PropertiesUtil.KAFKA_PROPERTIES.getProperty("kafka.data_access.topic");
    public static final String KAFKA_TEST_TOPIC = PropertiesUtil.KAFKA_PROPERTIES.getProperty("kafka.test.topic");

    @Override
    public void onMessage(ConsumerRecord<String, String> consumerRecord) {
        String topic = consumerRecord.topic();
        log.info("topic: " + topic);
        String value = consumerRecord.value();
        log.info("value: " + value);

        if (KAFKA_DATA_ACCESS_TOPIC.equals(topic)) {
            JSONObject.parseArray(value, DeviceOnlineData.class).stream().forEach(deviceOnlineData -> {
                try {
                    DATA_ACCESS_BLOCKING_QUEUE.put(deviceOnlineData);
                    log.info("Put into blocking queue");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        } else if (KAFKA_TEST_TOPIC.equals(topic)) {
            log.info("Test success");
        } else {
            log.info("Unknown topic");
        }
    }
}