简介
消息队列
为什么要有消息队列
图-1 消息队列的使用
消息队列
1)消息Message：网络中的两台计算机或者两个通讯设备之间传递的数据。例如说：文本、音乐、视频等内容。
2)队列Queue：一种特殊的线性表(数据元素首尾相接)，特殊之处在于只允许在首部删除元素和在尾部追加元素(FIFO)。入队、出队。
3)消息队列MQ：消息+队列，保存消息的队列。消息的传输过程中的容器；主要提供生产、消费接口供外部调用做数据的存储和获取。
消息队列的分类
MQ主要分为两类：点对点(p2p)、发布订阅(Pub/Sub)。
1)点对点模式(一对一，消费者主动拉取数据，消息收到后消息清除)消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。消息被消费以后，queue中不再有存储，所以消息消费者不可能消费到已经被消费的消息。Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
图-2 点对点模式
- 发布/订阅模式(一对多，消费者消费数据之后不会清除消息)消息生产者(发布)将消息发布到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/
- 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();
}
}
}
