Message Queue（MQ），消息队列中间件。很多人说:MQ 通过发送和接收消息来实现应用程序的异步和解偶，给人的直觉是——MQ 是异步解耦，但这只是 MQ 而不是目的。MQ 真正的目的是屏蔽底层复杂的通信协议，定义一套应用层更简单的通信协议。

• 传统定义：Kafka基于发布/订阅模式的分布式消息队列（Message Queue），主要用于大数据实时处理领域。

• 最新定义：Kafka是 开源分布式事件流平台 （Event Streaming Platform），数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用。

两种模式

作为消息队列，有两种基本操作模式：

• 点对点模式

消费者主动拉取数据，消费后删除

• 发布订阅模式

特点：可以有多个topic主题(浏览、喜欢、收集、评论等。).消费者消费数据后，不删除数据。每个消费者都是独立的，可以消费数据。

下载安装

首先进入官网下载：https://kafka.apache.org/downloads

安装后先启动zookeeper，在启动Kafka。（注意Kafka3.可以不用0开始zookeeper了）

首先是建立 topic(主题)在这里建立了一个test主题：

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

然后查看相应的主题：

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

删除主题命令：

kafka-topics --delete --zookeeper 127.0.0.1:2181 --topic test 

使用Kafka逆行生产者和消费者通信

kafka自带了一个producer命令客户端可以从当地文件中读取内容，也可以直接在命令行中输入内容，并以消息的形式发送到kafka集群。默认情况下，每个行会都被视为一个独立的消息。kafka客户端发送消息，指定发送kafka服务器地址及topic。

发送消息(生产者)的命令如下:

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

对于consumer，kafka还携带命令行客户端，将在命令中获取内容 出， 默认是消费的最新消息 。使用kafka从指定的消费者消息客户端kafka指定服务器 topic中消费消息

消费者信息(消费者)命令如下:

# 方法1：从最后一条消息的偏移量 1开始消费 kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --topic test  # 方法二:从零开始消费 kafka-console-consumer.bat --bootstrap-server 127.0.0.1:9092 --from-beginning --topic test 

效果如下：

首先，让我们来看看基本的新闻(Message)相关术语：

服务端(brokers)和客户端(producer、consumer)通过通信 TCP协议 来完成。

注意点

通过这些介绍，我们将解释上述测试中的一些说明：

• 新闻将被存储 （存储在log文件中）
• 信息是顺序存储(满足队列要求)
• 消息有偏移(比如消费消费指定从哪里读取，前者不会被消费)
• 消费时可以指出偏移量

细节方面

1. 发送消息的方会将消息发送到broker中，broker会存储消息，消息按发送顺序存储。因此，消费者可以在消费消息时指出主题中消息的偏移。默认情况下，消费始于最后一条消息的下一个偏移。

2. 单播新闻：消费 组 只有一个消费者可以消费某个消费者topic新闻。因此，可以在同一消费群体中创建多个消费者。

   kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --consumer-property group.id=testGroup --topic test 

   每个消费者都会执行这一点，并加入它 testGroup这个组中。

3. 多播实现： 在一些业务场景中需要让一条消息被多个消费者消费，那么就可以使用多播模式。

   kafka实现多播，只需要让不同的消费者处于不同的消费组即可。

   kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --consumer-property group.id=testGroup1 --topic test
   
   kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --consumer-property group.id=testGroup2 --topic test
   

4. 查看消费组以及消费信息

   # 查看当前主题下有哪些消费组
   kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --list
   
   # 查看消费组中的具体信息：比如当前偏移量、最后一条消息的偏移量、堆积的消息数量
   kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group testGroup
   

• Currennt-offset: 当前消费组的已消费偏移量

• Log-end-offset: 主题对应分区消息的结束偏移量(HW)

• Lag: 当前消费组未消费的消息数

kafka配置文件 server.properties

首先是这个部分的配置文件，相关配置：

主题没啥好说的了，就一个分类而已。

partition 分区

上面说过，一个主题只是一个逻辑的概念，实际的数据是存储在分区里面。

一个主题中的消息量是非常大的，因此可以通过分区的设置，来分布式存储这些消息。比如一个topic创建了 3 个分区。那么topic中的消息就会分别存放在这三个分区中。命令如下：

# 一个主题两个分区 --partitions 2 就表示两个分区
kafka-topics.sh --create --zookeeper localhost:2181 --partitions 2 --topic test1

# 查看分区的信息
kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1

分区的作用以及系统分区

第一点也就是最重要的一点，就是可以分布式的存储。什么概念，Kafka是可以集群的，当处于一个集群的时候。每个节点中相同的主题中分区就是数据共享的。也正是这样，就可以有了第二个点可以并发的写数据。

数据实际上存储在：data/kafka-logs/test-0（这个路径是Kafka配置文件指定）中的0000000.log文件中。

这里有个细节，就是明明我只创建了test主题并分配一个分区，按理来说只有test-0这个目录，但实际上这里生成了50个__consumer_offsets目录，这个其实就是自带的分区，用于记录消费者消费到了那个消息！

上面提到过偏移量概念，这里其实也是一个偏移量的概念。

总之，消费者消费消息，会提交自己消费到了哪里（也就是offset），这个就会保存到_consumer_offsets里面，这个保存的结构就是 k-v 结构， k e y = c o n s u m e r G r o u p I d + t o p i c + 分 区 号 ， v a l u e = o f f s e t key=consumerGroupId + topic + 分区号，value=offset key=consumerGroupId+topic+分区号，value=offset。

这样一来，同一个消费群组（consumerGroup）有一个挂了，另外的消费者就可以通过key找到是哪个__consumer_offsets记录topic中的分区，再去通过value找到消费到了那里，就可以继续消费了。

那么这里又有一个问题就是怎么知道消费者提交的 offset 存储到了哪个__consumer_offsets分区中呢？其实这里就是一个分布式系统常见的计算方式：hash(consumerGroupId) % __consumer_offsets主题的分区个数

集群的搭建

假设搭建三个节点的集群，所以需要准备三个server.properties配置文件。只需要更改以下内容：

# 节点一 由于这里都是在同一个机器上部署，所以listeners都是一个样的，只有后面的端口不一样
broker.id= 0
listeners=PLAINTEXT://127.0.0.1:9092
log.dir=/usr/local/data/kafka-logs

# 节点二 由于这里都是在同一个机器上部署，所以listeners都是一个样的，只有后面的端口不一样
broker.id= 1
listeners=PLAINTEXT://127.0.0.1:9093
log.dir=/usr/local/data/kafka-logs1

# 节点二 由于这里都是在同一个机器上部署，所以listeners都是一个样的，只有后面的端口不一样
broker.id= 2
listeners=PLAINTEXT://127.0.0.1:9094
log.dir=/usr/local/data/kafka-logs2

启动命令如下：

# 指定不同的配置文件即可 -daemon 的意思就是守护进程方式（也就是后台启动）
kafka-server-start.sh -daemon ../../config/server.properties

集群发送消息命令如下：

kafka-console-producer.sh --broker-list 127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094 --topic test

接收消费消息命令如下：

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094 --from-beginning --topic test

副本概念

副本机制是许多存储引擎必备的，在数据存储时候，为了保证数据的高可靠性，常常需要将主节点数据进行备份存储，即保存一份与主节点相同的数据集，一旦主节点发生宕机等故障，通过副本还能进行数据的恢复。

在集群中，不同的副本会被部署在不同的broker上。下面例子：创建 1个主题， 2 个分区、 3 个副本。

kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 3 --partitions 2 --topic test

根据这个集群，创建之后，三个节点都有这个主题，并且都有两个分区，其中只有两个分区是leader用来读写数据，另外的4个是这两个的备份。

使用如下命令查看主题的详细信息：

kafka-topics.bat --describe --zookeeper 127.0.0.1:2181 -topic test

通过查看主题信息，其中的关键数据：

• replicas：当前副本存在的broker节点

• leader：副本里的概念，这里的数字标识当前分区leader位于哪个broker

  每个partition都有一个broker作为leader。消息发送方要把消息发给哪个broker？就看副本的leader是在哪个broker上面。副本里的leader专⻔用来接收消息。接收到消息，其他follower通过poll的方式来同步数据。

• follower：leader处理所有针对这个partition的读写请求，而follower被动复制leader，不提供读写（主要是为了保证多副本数据与消费的一致性），如果leader所在的broker挂掉，那么就会进行新leader的选举。

• isr： 可以同步的broker节点和已同步的broker节点，存放在isr集合中。

总结几个点

• 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率；因此副本的数量要合理设置

• Kafka 中副本类型有2种，分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据；

• Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）；AR = ISR + OSR

  • ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。

  • OSR ， 表示 Follower 与 Leader 副本同步时，延迟过多的副本

根据这张图来进行说明：集群有两个broker，每个broker中有多个partition

• 一个partition只能被一个消费组里的某一个消费者消费，从而保证消费顺序。注意i这个保证是一个局部性的。

• Kafka只在partition的范围内保证消息消费的局部顺序性，不能在同一个topic中的多个partition中保证总的消费顺序性。一个消费者可以消费多个partition，只要这个partition没有被同一个组的其他人消费即可。

• 根据上面的机制其实可以知道：消费组中消费者的数量不能比一个topic中的partition数量多，否则多出来的消费者消费不到消息。

不论是生产者还是消费者，都需要引入依赖：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.4.1</version>
</dependency>

public class MyProducer { 
        

    private final static String TOPIC_NAME = "test";

    public static void main(String[] args) throws ExecutionException, InterruptedException { 
        
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, "test1", "value-test1");
        RecordMetadata metadata = producer.send(producerRecord).get();
        //=====阻塞=======
        System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
    }
}

指定分区发送

ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0 , "key", "value");

未指定分区

如果不指定分区，则会通过业务key的hash运算，算出消息往哪个分区上发

具体发送的分区计算公式：hash(key) % partitionNum

生产者ACK配置

在同步发消息的场景下：生产者发动broker上后，ack会有 3 种不同的选择：

• acks=0：表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
• acks=1：至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
• acks=-1或all：需要等待 min.insync.replicas(默认为 1 ，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。

需要在设置属性的时候设置：

props.put(ProducerConfig.ACKS_CONFIG, "1");

生产者发消息，发送完后不用等待broker给回复，直接执行下面的业务逻辑。可以提供callback，让broker异步的调用callback，告知生产者，消息发送的结果。

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
//把发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//把发送消息value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

Producer<String, String> producer = new KafkaProducer<String, String>(props);
//指定发送分区
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, 0, "key1", "value1");
//异步回调方式发送消息
producer.send(producerRecord, new Callback() { 
        
    public void onCompletion(RecordMetadata metadata, Exception exception) { 
        
        if (exception != null) { 
        
            System.err.println("发送消息失败：" +
                               exception.getStackTrace());
        }
        if (metadata != null) { 
        
            System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-" + metadata.partition() + "|offset-" + metadata.offset());
        }
    }
});

其他细节

• 发送会默认会重试 3 次，每次间隔100ms
• 发送的消息会先进入到本地缓冲区（32mb），kakfa会跑一个线程，该线程去缓冲区中取16k的数据，发送到kafka，如果到 10 毫秒数据没取满16k，也会发送一次。

代码如下：

private final static String TOPIC_NAME = "test";
private final static String CONSUMER_GROUP_NAME = "testGroup";

public static void main(String[] args) { 
        
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
    // 消费分组名
    props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    //创建一个消费者的客户端
    KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    // 消费者订阅主题列表
    consumer.subscribe(Arrays.asList(TOPIC_NAME));

    while (true) { 
        
        /* * poll() API 是拉取消息的⻓轮询 */
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) { 
        
            System.out.printf("收到消息：partition = %d,offset = %d, key =%s, value = %s%n", record.partition(), record.offset(), record.key(), record.value());
        }
    }
}

• 设置自动提交参数 - 默认

// 是否自动提交offset，默认就是true
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
// 自动提交offset的间隔时间
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

消费者poll到消息后默认情况下，会自动向broker的_consumer_offsets主题提交当前主题-分区消费的偏移量。

自动提交会丢消息： 因为如果消费者还没消费完poll下来的消息就自动提交了偏移量，那么此 时消费者挂了，于是下一个消费者会从已提交的offset的下一个位置开始消费消息。之前未被消费的消息就丢失掉了。

• 设置手动提交参数

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

在消费完成之后进行提交：（手动同步提交）

if (records.count() > 0 ) { 
        
    // 手动同步提交offset，当前线程会阻塞直到offset提交成功
    // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
    consumer.commitSync();
}

手动异步提交：

if (records.count() > 0 ) { 
        
    // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
    consumer.commitAsync(new OffsetCommitCallback() { 
        
        @Override
        public void onComplete(Map<TopicPartition, OffsetAndMetadata>offsets, Exception exception) { 
        
            if (exception != null) { 
        
                System.err.println("Commit failed for " + offsets);
                System.err.println("Commit failed exception: " +exception.getStackTrace());
            }
        }
    });
}

其他细节