消息队列底子概念及选型，常见解决方案包罗消息可靠性、消息有序、消息堆积、重复斲丧、变乱消息

前言

是时间总结下消息队列干系知识点啦！我搓搓搓搓
本文包罗消息队列底子概念先容，常见解决方案包罗消息可靠性、消息有序、消息堆积、重复斲丧、变乱消息
参考资料：
Kafka常见标题总结 | JavaGuide
RocketMQ常见标题总结 | JavaGuide
【原创】消息队列的斲丧语义和投递语义 - 孤独烟 - 博客园
Kafka变乱是怎么实现的？Kafka变乱消息原理详解（文末送书）-CSDN博客
Kafka/RocketMQ变乱消息对比 - 简书
1. 什么是消息队列

消息队列可以看作是存放消息的容器，常用于分布式体系中，在消息的生产者和斲丧者之间引入一个缓冲区
它的作用紧张由三点概括，分别是

• 解耦：生产者和斲丧者之间没有直接的调用关系，新增大概修改模块对别的模块的影响较小
  
• 异步：生产者无需期待消息斲丧完成绩马上返回，淘汰用户哀求的相应时间
  
• 削峰：在高并发场景，消息队列可以缓存消息，平滑高峰流量，防止体系过载
  

2. 消息队列有什么常见应用场景

日志处理处罚：将日志发送到消息队列，由日志处理处罚体系举行斲丧，举行及时分析和监控
电商体系：订单的创建、付出、发货等步调可以由消息队列举行异步处理处罚息争耦
任务调理：任务调理体系可以将任务发布在消息队列中，由多个差别的节点并行处理处罚
数据同步：消息队列可以将变更的数据同步到差别的存储体系中
3. 常见的消息队列怎样选型

常见的消息队列有几种：RabbitMQ、Kafka、RocketMQ
从延时来看，RabbitMQ是基于Erlang开辟的，延时最低，能到达微秒级别，别的都是毫秒级别
从吞吐量来看，Kafka和RocketMQ吞吐量最高，达十万、百万级，RabbitMQ只有万级
从定制开辟难度来看，RocketMQ是由Java开辟的，对于大型公司可以有人手举行定制化开辟，而能对RabbitMQ举行定制化开辟的较少，但是RabbitMQ社区生动，可以或许办理开辟上的bug
从业务场景来看，RocketMQ根本上担当住了大型企业场景的磨练好比双十一，Kafka实用于大数据场景及时分析和日志收罗等业务
总结，如果是数据量没有那么大，可以选择功能完备、社区生动的RabbitMQ，如果有大数据量的金融互联网场景，对可靠性和吞吐量要求都很高，保举选择RocketMQ，如果是日志收罗和大数据及时分析的场景，则保举使用Kafka
4. 消息队列有什么模子

JMS是一种Java消息服务的API规范，它界说了两种消息模子：队列模子和发布/订阅模子
队列模子又叫点对点模子，指的是生产者往队列里发送消息，一个消息只能被一个斲丧者斲丧，一个队列的多个斲丧者之间是竞争关系
发布/订阅模子又叫主题模子，指生产者往Topic发送消息，全部订阅了该Topic的斲丧者都能斲丧到
Kafka和RocketMQ就是基于发布/订阅模子实现的
AMQP是另一种消息服务的协议，它界说了几种消息模子，通过引入exchange完成对消息的路由，支持direct exchange、fanout exchange、topic exchange
RabbitMQ就是基于AMQP协议实现的
5. 怎样包管消息不丢失

消息不丢失必要从生产消息、存储消息、斲丧消息三个方面去包管
5.1 生产消息

在生产者侧，必要处理处罚好生产消息的非常处理处罚，如果写入失败必要有重试、告警等机制
5.2 存储消息

在存储消息上，Broker必要在刷盘之后再给生产者应答
在RocketMQ中，可以设置同步刷盘，主从架构下设置同步复制（也叫同步双写）
在Kafka中，默认acks=1，表现消息被leader副本担当后就会返回乐成，设置acks=all，那么在全部的ISR(In-Sync Replicas)担当后才算乐成，干系的别的参数有：
replication.factor：分区的副本数，可设置为>=3
min.insync.replicas：消息发送乐成必要至少写入的副本数，设置为>1
unclean.leader.election.enable：leader副本故障后的推举机制，false表现不从非ISR节点中选leader，设置为false
5.3 斲丧消息

斲丧者在真正完成业务逻辑时再返回乐成给Broker
在RocketMQ中，完成斲丧逻辑后再返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS，否则返回ConsumeConcurrentlyStatus.RECONSUME_LATER

1. import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
2. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
3. import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
4. import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
5. import org.apache.rocketmq.common.message.MessageExt;
6. import java.util.List;
8. public class RocketMQConsumer {
9.     public static void main(String[] args) throws Exception {
10.         // 实例化消费者，并指定消费者组名
11.         DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_consumer_group");
12.         // 指定NameServer地址
13.         consumer.setNamesrvAddr("localhost:9876");
14.         // 订阅一个或多个Topic，以及Tag来过滤需要消费的消息
15.         consumer.subscribe("TopicTest", "*");
16.         // 注册消息监听器
17.         consumer.registerMessageListener(new MessageListenerConcurrently() {
18.             @Override
19.             public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
20.                 for (MessageExt msg : msgs) {
21.                     try {
22.                         // 处理消息的业务逻辑
23.                         System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));
24.                     } catch (Exception e) {
25.                         // 处理异常，这里可以根据业务需求进行重试或其他处理
26.                         e.printStackTrace();
27.                         // 消费失败，返回RECONSUME_LATER，消息会重新投递
28.                         return ConsumeConcurrentlyStatus.RECONSUME_LATER;
29.                     }
30.                 }
31.                 // 消费成功，返回CONSUME_SUCCESS
32.                 return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
33.             }
34.         });
35.         // 启动消费者实例
36.         consumer.start();
37.         System.out.printf("Consumer Started.%n");
38.     }
39. }   

复制代码

在Kafka中，必要设置enable.auto.commit=false，然后在代码中手动提交offset

1. import org.apache.kafka.clients.consumer.*;
2. import org.apache.kafka.common.TopicPartition;
4. import java.time.Duration;
5. import java.util.Collections;
6. import java.util.HashMap;
7. import java.util.Map;
8. import java.util.Properties;
10. public class KafkaConsumerExample {
11.     public static void main(String[] args) {
12.         // 配置消费者属性
13.         Properties props = new Properties();
14.         props.put("bootstrap.servers", "localhost:9092");
15.         props.put("group.id", "test-group");
16.         props.put("enable.auto.commit", "false");
17.         props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
18.         props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
20.         // 创建Kafka消费者实例
21.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
23.         // 订阅主题
24.         String topic = "test-topic";
25.         consumer.subscribe(Collections.singletonList(topic));
27.         try {
28.             while (true) {
29.                 // 拉取消息
30.                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
31.                 Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
32.                 for (ConsumerRecord<String, String> record : records) {
33.                     try {
34.                         // 处理消息的业务逻辑
35.                         System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
36.                         // 记录需要提交的偏移量
37.                         TopicPartition partition = new TopicPartition(record.topic(), record.partition());
38.                         OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(record.offset() + 1);
39.                         offsetsToCommit.put(partition, offsetAndMetadata);
40.                     } catch (Exception e) {
41.                         // 处理异常，这里可以根据业务需求进行重试或其他处理
42.                         e.printStackTrace();
43.                     }
44.                 }
45.                 // 手动同步提交偏移量
46.                 if (!offsetsToCommit.isEmpty()) {
47.                     consumer.commitSync(offsetsToCommit);
48.                 }
49.             }
50.         } finally {
51.             // 关闭消费者
52.             consumer.close();
53.         }
54.     }
55. }   

复制代码

6. 怎样办理重复斲丧

办理重复斲丧比力粗鲁的办法就是包管一条消息只能被斲丧一次，但是如果斲丧者挂了未提交offset，那么为了包管消息可靠性，消息就会被重复斲丧
以是重点是要让斲丧者的处理处罚具有幂等性，即多次处理处罚同一条消息得到的结果是一样的
详细做法有引入全局唯一的ID，对已经斲丧过的消息举行去重
7. 怎样包管消息的有序性

简朴粗暴的方法是：单一生产者和单一斲丧者，即消息只由单个生产者发往单个队列，再由单个斲丧者斲丧，存在性能瓶颈
在支持分区键（Partition Key）的体系好比Kafka和RocketMQ中，可以在发送消息的时间指定key
在Kafka中在发送消息时指定key，如果必要严酷序次的话，可以指定max.in.flight.requests.per.connection=1，该参数表现在得到相应前可以发送的消息数，参数值越大吞吐量越大，设置为1会低落吞吐量
在RocketMQ中可以通过继承MessageQueueSelector实现

1. import org.apache.rocketmq.client.exception.MQClientException;
2. import org.apache.rocketmq.client.producer.DefaultMQProducer;
3. import org.apache.rocketmq.client.producer.MessageQueueSelector;
4. import org.apache.rocketmq.client.producer.SendResult;
5. import org.apache.rocketmq.common.message.Message;
6. import org.apache.rocketmq.common.message.MessageQueue;
8. import java.util.List;
10. public class RocketMQPartitionKeyProducer {
11.     public static void main(String[] args) throws MQClientException {
12.         DefaultMQProducer producer = new DefaultMQProducer("partition_key_producer_group");
13.         producer.setNamesrvAddr("localhost:9876");
14.         producer.start();
16.         String topic = "partition_key_topic";
17.         // 模拟不同的 Partition Key
18.         String[] partitionKeys = {"key1", "key2", "key1", "key2"};
19.         for (int i = 0; i < partitionKeys.length; i++) {
20.             String partitionKey = partitionKeys[i];
21.             Message msg = new Message(topic, "TAG", ("Message " + i).getBytes());
22.             try {
23.                 SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
24.                     @Override
25.                     public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
26.                         String key = (String) arg;
27.                         int index = Math.abs(key.hashCode()) % mqs.size();
28.                         return mqs.get(index);
29.                     }
30.                 }, partitionKey);
31.                 System.out.printf("%s%n", sendResult);
32.             } catch (Exception e) {
33.                 e.printStackTrace();
34.             }
35.         }
36.         producer.shutdown();
37.     }
38. }

复制代码

8. 推拉模式有什么区别，怎么选择

推模式和拉模式的选择一样寻常出如今Broker和斲丧者之间，而生产者和Broker之间一样寻常使用推模式而不是Broker去拉取消息，否则大量的生产者还必要去维护消息可靠性
推模式的利益是Broker在吸收到消息后可以或许敏捷地推送给斲丧者举行斲丧，实用于对及时性要求比力高的场景，弊端是斲丧者斲丧本领不敷强时容易过载，必要Broker去维护斲丧者的状态去调解推送速率，对Broker要求比力高
拉模式的利益是对Broker没什么要求，可以稳固地控制消息斲丧的速率，防止斲丧者过载，而且得当批量地拉取消息，弊端是消息容易耽误，大概是长时间轮询没有消息可斲丧
RocketMQ和Kafka都使用了拉模式，它们都使用长轮询对拉模式做了优化，也就是在斲丧者拉取的时间把哀求给hold住，然后期待消息到来再把消息发送出去
9. 怎样办理消息堆积

消息堆积标题有两个缘故原由，一个是生产太快，一个是斲丧太慢
生产太快的标题可以对生产端做限流和降级，以及只生存关键消息，抛弃非关键消息，大概耽误处理处罚
斲丧太慢的标题必要去定位bug，提拔斲丧者的斲丧本领，包罗优化斲丧的逻辑，增长斲丧者的线程数量，对斲丧者举行垂直扩容（增长单个斲丧者的CPU/内存），大概程度扩容（增长斲丧者副本数）
10. Kafka的变乱消息是什么，怎么实现

Kafka的变乱消息包管一系列消息要么完全发送乐成，要么完全发送失败，即一系列消息操纵的原子性
实现的方式是，Kafka使用变乱调和器负责变乱的启动、提交和制止，为确保准确一次（exactly once）的投递语义，会为每个Producer维护一个pid，为<pid, topic, partition>维护一个递增的seq从而包管消息按序次被准确吸收，在斲丧者侧，必要设置read_committed=true包管只斲丧已提交的消息
详细实践如下
生产者设置acks=all确认消息乐成发送、transaction_id设置变乱id、enable.idempotence=true确保投递准确一次

1. import org.apache.kafka.clients.producer.KafkaProducer;
2. import org.apache.kafka.clients.producer.ProducerConfig;
3. import org.apache.kafka.clients.producer.ProducerRecord;
4. import org.apache.kafka.common.serialization.StringSerializer;
5. import java.util.Properties;
7. public class KafkaTransactionalProducer {
8.     public static void main(String[] args) {
9.         Properties props = new Properties();
10.         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
11.         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
12.         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
13.         props.put(ProducerConfig.ACKS_CONFIG, "all");
14.         props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
15.         props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
17.         KafkaProducer<String, String> producer = new KafkaProducer<>(props);
18.         producer.initTransactions();
20.         try {
21.             producer.beginTransaction();
22.             ProducerRecord<String, String> record = new ProducerRecord<>("test_topic", "key", "value");
23.             producer.send(record);
24.             producer.commitTransaction();
25.         } catch (Exception e) {
26.             producer.abortTransaction();
27.             e.printStackTrace();
28.         } finally {
29.             producer.close();
30.         }
31.     }
32. }

复制代码

斲丧者设置read_committed=true确保只斲丧已提交的消息，auto.offset.reset=earliest确保启动时从最早的消息开始斲丧

1. import org.apache.kafka.clients.consumer.ConsumerConfig;
2. import org.apache.kafka.clients.consumer.ConsumerRecords;
3. import org.apache.kafka.clients.consumer.KafkaConsumer;
4. import org.apache.kafka.common.serialization.StringDeserializer;
5. import java.time.Duration;
6. import java.util.Collections;
7. import java.util.Properties;
9. public class KafkaTransactionalConsumer {
10.     public static void main(String[] args) {
11.         Properties props = new Properties();
12.         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
13.         props.put(ConsumerConfig.GROUP_ID_CONFIG, "test_group");
14.         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
15.         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
16.         props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
17.         props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
19.         KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
20.         consumer.subscribe(Collections.singletonList("test_topic"));
22.         while (true) {
23.             ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
24.             for (var record : records) {
25.                 System.out.printf("Received message: key = %s, value = %s%n", record.key(), record.value());
26.             }
27.         }
28.     }
29. }

复制代码

11. RocketMQ的变乱消息是什么

RocketMQ的变乱消息和Kafka的差别，Kafka实现的是多条消息的原子性，RocketMQ实现的是分布式变乱，也就是消息存储和当地变乱处在同一个变乱
RocketMQ是使用变乱消息+变乱回查机制实现分布式变乱的

• 生产者发送“半消息”到MQ
  
• MQ相应
  
• 生产者实验当地变乱
  
• 生产者根据当地变乱实验状态向MQ发送commit/rollback，如果commit，半消息变为正式消息，斲丧者可见，如果rollback，则抛弃该半消息
  
• 如果由于网络抖动等缘故原由，MQ充公到第四步，则关照生产者变乱回查
  
• 生产者查询变乱状态
  
• 生产者根据回查的变乱状态，关照MQ举行commit/rollback