简介

消息队列

为什么要有消息队列

图-1 消息队列的使用

消息队列

1)消息Message：网络中的两台计算机或者两个通讯设备之间传递的数据。例如说：文本、音乐、视频等内容。

2)队列Queue：一种特殊的线性表(数据元素首尾相接)，特殊之处在于只允许在首部删除元素和在尾部追加元素(FIFO)。入队、出队。

3)消息队列MQ：消息+队列，保存消息的队列。消息的传输过程中的容器；主要提供生产、消费接口供外部调用做数据的存储和获取。

消息队列的分类

MQ主要分为两类：点对点(p2p)、发布订阅(Pub/Sub)。

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

图-2 点对点模式

1. 发布/订阅模式(一对多，消费者消费数据之后不会清除消息)消息生产者(发布)将消息发布到topic中，同时有多个消息消费者(订阅)消费该消息。和点对点方式不同，发布到topic的消息会被所有订阅者消费。消息可以传给多个消费者。

图-3 发布订阅模式

p2p和发布订阅MQ的比较

共同点

消息生产者生产消息发送到queue中，然后消息消费者从queue中读取并且消费消息。

不同点

1)p2p模型包括：消息队列(Queue)、发送者(Sender)、接收者(Receiver)

一个生产者生产的消息只有一个消费者(Consumer)(即一旦被消费，消息就不在消息队列中)。比如说打电话。

2)pub/Sub包含：消息队列(Queue)、主题(Topic)、发布者(Publisher)、订阅者(Subscriber)。

3)每个消息可以有多个消费者，彼此互不影响。比如我发布一个微博：关注我的人都能够看到。

消息系统的使用场景

1)解耦：各系统之间通过消息系统这个统一的接口交换数据，无须了解彼此的存在。

2)冗余：部分消息系统具有消息持久化能力，可规避消息处理前丢失的风险。 3)扩展：消息系统是统一的数据接口，各系统可独立扩展。

4)峰值处理能力：消息系统可顶住峰值流量，业务系统可根据处理能力从消息系统中获取并处理对应量的请求。

5)可恢复性：系统中部分键失效并不会影响整个系统，它恢复会仍然可从消息系统中获取并处理数据。

6)异步通信：在不需要立即处理请求的场景下，可以将请求放入消息系统，合适的时候再处理。

常见的消息系统

1)RabbitMQ Erlang编写，支持多协议AMQP，XMPP，SMTP，STOMP。支持负载均衡、数据持久化。同时支持Peer-to-Peer和发布/订阅模式。

2)Redis 基于Key-Value对的NoSQL数据库，同时支持MQ功能，可做轻量级队列服务使用。就入队操作而言，Redis对短消息(小于10kb)的性能比RabbitMQ好，长消息性能比RabbitMQ差。

3)ZeroMQ 轻量级，不需要单独的消息服务器或中间件，应用程序本身扮演该角色，Peer-to-Peer。它实质上是一个库，需要开发人员自己组合多种技术，使用复杂度高。

4)ActiveMQ JMS实现，Peer-to-Peer，支持持久化、XA(分布式)事务。

5)Kafka/Jafka 高性能跨语言的分布式发布/订阅消息系统，数据持久化，全分布式，同时支持在线和离线处理。

6)MetaQ/RocketMQ 纯Java实现，发布/订阅消息系统，支持本地事务和XA分布式事务。

Kafka简介

Kafka概述

Kafka是分布式的发布—订阅消息系统。它最初由LinkedIn(领英)公司发布，使用Scala语言编写，于2010年12月份开源，成为Apache的顶级项目。Kafka是一个高吞吐量的、持久性的、分布式发布订阅消息系统。它主要用于处理活跃live的数据(登录、浏览、点击、分享、喜欢等用户行为产生的数据)。如图-4所示，很好的显示了Kafka的应用与组成。

图-4 kafka应用与组成

特点

1)高吞吐量：可以满足每秒百万级别消息的生产和消费——生产消费。

2)持久性：有一套完善的消息存储机制，确保数据的高效安全的持久化——中间存储。

3)分布式：基于分布式的扩展和容错机制；Kafka的数据都会复制到几台服务器上。当某一台故障失效时，生产者和消费者转而使用其它的机器——整体。

4)健壮性。

设计目标

1)高吞吐率：在廉价的商用机器上单机可支持每秒100万条消息的读写。

2)消息持久化：所有消息均被持久化到磁盘，无消息丢失，支持消息重放。

3)完全分布式：Producer，Broker，Consumer均支持水平扩展。

4)同时适应在线流处理和离线批处理。

Kafka核心概念

概述

一个MQ需要哪些部分？生产、消费、消息类别、存储等等。

对于kafka而言，kafka服务就像是一个大的水池。不断的生产、存储、消费着各种类别的消息。那么kafka由何组成呢？

Kafka服务

1)Topic：主题，Kafka处理的消息的不同分类。

2)Broker：消息服务器代理，Kafka集群中的一个kafka服务节点称为一个broker，主要存储消息数据。存在硬盘中。每个topic都是有分区的。

3)Partition：Topic物理上的分组，一个topic在broker中被分为1个或者多个partition，分区在创建topic的时候指定。

4)Message：消息，是通信的基本单位，每个消息都属于一个partition。

Kafka服务相关

1)Producer：消息和数据的生产者，向Kafka的一个topic发布消息。

2)Consumer：消息和数据的消费者，定于topic并处理其发布的消息。

3)Zookeeper：协调kafka的正常运行。

Kafka安装与操作

安装与配置

版本说明

安装包下载地址：

Index of /dist/kafka/3.5.0

源码包下载地址：

Index of /dist/kafka/3.5.0

安装配置

1)解压：

tar -zxvf kafka_2.12-3.5.0.tgz -C /opt/module/

2)重命名：

mv kafka_2.12-3.5.0/ kafka-3.5.0/

1. hadoop101执行以下命令创建数据文件存放目录：

mkdir -p /opt/module/kafka-3.5.0/datas

4)修改配置文件：

// 修改$KAFKA_HOME/config/server.properties

// 当前kafka实例的id，必须为整数，一个集群中不可重复

broker.id=0

// 生产到kafka中的数据存储的目录,目录需要手动创建

log.dirs=/opt/module/kafka-3.5.0/datas

// kafka数据在zk中的存储目录

zookeeper.connect=hadoop101:2181,hadoop102:2181,hadoop103:2181/kafka

// 添加配置，用来删除topic

delete.topic.enable=true

host.name=hadoop101

5)同步到其他机器：

scp -r kafka/ hadoop102:$PWD

scp -r kafka/ hadoop103:$PWD

6)修改broker.id：

//修改broker.id

broker.id=1

broker.id=2

//修改host.name

host.name=hadoop102

host.name=hadoop103

服务启动

服务启动:每台都要运行此命令：

nohup bin/kafka-server-start.sh config/server.properties 2>&1 &

启动结果如图-5所示：

图-5 kafka启动

Kafka基本操作

Kafka的topic操作

topic是kafka非常重要的核心概念，是用来存储各种类型的数据的，所以最基本的就需要学会如何在kafka中创建、修改、删除的topic，以及如何向topic生产消费数据。

关于topic的操作脚本：kafka-topics.sh：

bin/kafka-topics.sh --create \

--topic hadoop \ // 指定要创建的topic的名称

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092\

//指定kafka关联的zk地址

--partitions 3 \ //指定该topic的分区个数

--replication-factor 3 //指定副本因子

创建topic

注意：指定副本因子的时候，不能大于broker实例个数，否则报错，如图-6所示：

图-6 kafka创建topic

当使用正确的方式，即将replication-factor设置为3，之后执行脚本命令，创建topic成功，如图-7所示。

图-7 zookeeper中的topic列表

与此同时，在kafka数据目录data.dir=/opt/module/kafka-3.5.0/datas/中有了新变化，如图-8所示。

图-8 kafka数据目录

查看topic列表

bin/kafka-topics.sh --list \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092

执行结果如图-9所示。

图-9 kafka topic列表

查看每一个topic的信息

bin/kafka-topics.sh --describe \

--topic hadoop \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092

执行结果如图-10所示。

图-10 kafka topic信息

其中partition，replicas，leader，isr代表的是什么意思呢。

1)Partition：当前topic对应的分区编号。

2)Replicas：副本因子，当前kafka对应的partition所在的broker实例的broker.id的列表。

3)Leader：该partition的所有副本中的leader领导者，处理所有kafka该partition读写请求。

4)ISR：该partition的存活的副本对应的broker实例的broker.id的列表。

修改一个topic

[root@hadoop101 kafka]$ bin/kafka-topics.sh --alter \

--topic hadoop \

--partitions 4 \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092

执行结果如图-11所示，可以看出partition由原先的3个变成了4个。

图-11 kafka topic修改partition

但是注意：partition个数，只能增加，不能减少，如图-12所示。

图-12 kafka topic partition只能增加

删除一个topic

[root@hadoop101 kafka]$ bin/kafka-topics.sh --delete \

--topic test \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092

执行结果如图-13所示

图-13 kafka topic删除

Kafka终端数据生产与消费

脚本简介

在$KAFKA_HOME/bin目录下面提供了很多脚本，其中kafka-console-producer.sh和kafka-console-consumer.sh分别用来在终端模拟生产和消费数据，即作为kafka topic的生产者和消费者存在。

生产数据

生产数据，执行以下的命令：

[root@hadoop101 kafka]$ bin/kafka-console-producer.sh \

--topic hadoop \

--bootstrap-server

 hadoop101:9092,hadoop102:9092,hadoop103:9092

如图-14所示：

图-14 kafka-console-producer生产数据

消费数据

类似的，消费刚刚生产的数据需要执行以下命令：

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh \

--topic hadoop \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092

如图-15所示：

图-15 kafka-console-producer生产数据

但遗憾的是，我们并没有看到刚刚生产的数据，这很好理解，比如新闻联播每晚7点开始了，结果你7点15才打开电视看新闻，自然7点到7点15之间的新闻你就会错过，如果你想要看这之间的新闻，那么就需要其提供回放的功能，幸运的是kafka不仅提供了从头开始回放数据的功能，还可以做到从任意的位置开始回放或者读取数据，这点功能是非常强大的。

那么此时重新在生产端生产数据，比如4,5,6，再看消费端，如图-16所示，就可以看到有数据产生了。

图-16 kafka-console-consumer消费数据

那么我想要读取1,2,3的数据，那该怎么办呢？此时只需要添加一个参数--from-beginning从最开始读取数据即可，如图-17所示：

图-17 kafka-console-consumer从头消费数据

Kafka的数据消费的总结

消费者与分区之间的关系

kafka消费者在消费数据的时候，都是分组别的。不同组的消费不受影响，相同组内的消费，需要注意，如果partition有3个，消费者有3个，那么便是每一个消费者消费其中一个partition对应的数据；如果有2个消费者，此时一个消费者消费其中一个partition数据，另一个消费者消费2个partition的数据。如果有超过3个的消费者，同一时间只能最多有3个消费者能消费得到数据，如图-18所示。

图-18 kafka消费数据的特点

如下命令查看不同分区中产生的数据：

第一个消费者：

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh \

--topic hadoop \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092 \

--partition 0 \

--offset earliest

第二个消费者：

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh \

--topic hadoop \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092 \

--partition 1 \

--offset earliest

第三个消费者：

[root@hadoop102 kafka]$ bin/kafka-console-consumer.sh \

--topic hadoop \

--bootstrap-server hadoop101:9092,hadoop102:9092,hadoop103:9092 \

--partition 2 \

--offset earliest

offset：是kafka的topic中的partition中的每一条消息的标识，如何区分该条消息在kafka对应的partition的位置，就是用该偏移量。offset的数据类型是Long，8个字节长度。offset在分区内是有序的，分区间是不一定有序。如果想要kafka中的数据全局有序，就只能让partition个数为1，如图-19所示。

图-19 kafka offset概念

在组内，kafka的topic的partition个数，代表了kafka的topic的并行度，同一时间最多可以有多个线程来消费topic的数据，所以如果要想提高kafka的topic的消费能力，应该增大partition的个数。

Kafka编程api

生产与消费API操作

创建Kafka项目

指定项目存储位置和maven坐标，如图-20所示

图-20 创建maven项目

指定maven依赖信息：

代码如下：

<dependencies>

        <dependency>

            <groupId>org.apache.kafka</groupId>

            <artifactId>kafka-clients</artifactId>

            <version>3.5.0</version>

        </dependency>

</dependencies>

Kafka生产者的api操作

代码如下：

public class OrderProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);

        for (int i = 0;i<100;i++){
        kafkaProducer.send(new ProducerRecord<String, String>("test","这是第"+i+"条数据"));
        }
        kafkaProducer.close();
    }
}

Kafka消费者api

代码如下：

public class OrderConsumer {

public static void main(String[] args) {

// 1\连接集群

Properties props = new Properties();

props.put("bootstrap.servers", "hadoop101:9092");

props.put("group.id", "test");

//以下两行代码 ---消费者自动提交offset值

props.put("enable.auto.commit", "true");

props.put("auto.commit.interval.ms",  "1000");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");



KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
kafkaConsumer.subscribe(Arrays.asList("test"));

while (true) {
    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        String value = consumerRecord.value();
        int partition = consumerRecord.partition();
        long offset = consumerRecord.offset();
        String key = consumerRecord.key();
        System.out.println("key:" + key + "value:" + value + "partition:" + partition + "offset:" + offset);
    }
}

}

}

指定分区数据进行消费

1)如果进程正在维护与该分区关联的某种本地状态(如本地磁盘上的键值存储)，那么它应该只获取它在磁盘上维护的分区的记录。

2)如果进程本身具有高可用性，并且如果失败则将重新启动(可能使用YARN，Mesos或AWS工具等集群管理框架，或作为流处理框架的一部分)。在这种情况下，Kafka不需要检测故障并重新分配分区，因为消耗过程将在另一台机器上重新启动。

public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);

        TopicPartition topicPartition = new TopicPartition("test", 0);
        TopicPartition topicPartition1 = new TopicPartition("test", 1);

        kafkaConsumer.assign(Arrays.asList(topicPartition, topicPartition1));

        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                String value = consumerRecord.value();
                int partition = consumerRecord.partition();
                long offset = consumerRecord.offset();
                String key = consumerRecord.key();
                System.out.println("key:" + key + "value:" + value + "partition:" + partition + "offset:" + offset);
            }
            kafkaConsumer.commitSync();
}

}

}