本文是学习Java多线程与高并发知识时做的笔记。

这部分内容比较多&＃xff0c;按照内容分为5个部分&＃xff1a;

1. 多线程基础篇
2. JUC篇
3. 同步容器和并发容器篇
4. 线程池篇
5. MQ篇

本篇为MQ篇。

目录

1 MQ

1.1 什么是MQ&＃xff1f;

1.2 为什么要使用MQ&＃xff1f;

1.2.1 传统的http请求有哪些缺点&＃xff1f;

1.2.2 多线程异步任务处理

1.2.3 MQ异步任务处理

1.3 MQ核心概念

1.4 MQ的简单实现

1.4.1 Maven依赖

1.4.2 多线程实现本地MQ

1.5 当前的主流MQ

1.5.1 ActiveMQ

1.5.2 Kafka

1.5.3 RocketMQ

1.5.4 RabbitMQ

1 MQ

1.1 什么是MQ&＃xff1f;

MQ&＃xff0c;message queue&＃xff0c;表示消息队列。通过典型的生产者和消费者模型&＃xff0c;生产者不断地向消息队列中生产消息&＃xff0c;消费者不断地从队列中获取消息。因为消息的生产和消费都是异步的&＃xff0c;而且只关心消息的发送和接收&＃xff0c;没有业务逻辑的侵入&＃xff0c;轻松地实现系统间解耦。又称为消息中间件&＃xff0c;通过利用高效可靠的消息传递机制进行平台无关的数据交流&＃xff0c;并基于数据通信来进行分布式系统的集成。

1.2 为什么要使用MQ&＃xff1f;

1.2.1 传统的http请求有哪些缺点&＃xff1f;

1) http请求基于request-response模型&＃xff0c;在高并发的情况下&＃xff0c;客户端向服务端发送大量请求&＃xff0c;可能导致服务端请求堆积。在请求堆积过多的情况下&＃xff0c;可能会造成服务器崩溃。

一般会在服务端入口处实现限流、整合服务以保护框架。

2) 在http请求的处理逻辑比较耗时的情况下&＃xff0c;可能造成客户端响应超时。所以耗时的业务逻辑尽量交给多线程或者MQ处理。

1.2.2 多线程异步任务处理

某个电商平台在新用户申请注册时&＃xff0c;客户端发送请求到服务器端&＃xff0c;服务器会执行一系列操作&＃xff1a;

1. 插入会员信息&＃xff0c;需要1s时间
2. 发送登录短信提醒&＃xff0c;需要3s时间
3. 发送新人优惠券&＃xff0c;需要3s时间

服务器端响应客户端请求一共需要7s时间&＃xff0c;在这7s内客户端阻塞&＃xff0c;影响到用户体验。

如果使用多线程进行异步处理&＃xff1a;

向数据库中插入会员数据后&＃xff0c;单独开启一个线程发送登录短信和新人优惠券。

服务器端响应客户端请求只需要1s时间。

但这么做的缺点是&＃xff1a;单独开启线程会消耗服务器CPU资源&＃xff0c;在处理大量请求时&＃xff0c;服务器压力巨大。

1.2.3 MQ异步任务处理

如果使用MQ进行异步处理&＃xff1a;

先向数据库中插入会员数据&＃xff0c;然后向MQ投递消息&＃xff0c;MQ服务器端将消息推送给消息的消费者后&＃xff0c;由消费者发送登录短信和新人优惠券。【异步、解耦】

服务器端响应客户端请求只需要1s时间&＃xff0c;而且不消耗CPU资源。

MQ如何实现抗高并发&＃xff1f;

MQ消费者根据自身的能力情况&＃xff0c;拉取MQ服务器端的消息进行处理。

这种 量力而行 有效地提升了消息中间件的抗高并发能力&＃xff0c;但这么做的缺点是可能会造成消息处理延迟。【削峰】

提升MQ消费者任务处理速度的方法&＃xff1a;

• 搭建消费者集群提升处理速率
• 设置消费者批量获取消息(默认情况下&＃xff0c;消费者一次获取一条消息)

1.3 MQ核心概念

Message&＃xff1a;消息&＃xff0c;应用之间传送的数据&＃xff0c;比较常用的数据格式是JSON。

Broker&＃xff1a;MQ服务器端&＃xff0c;提供了消息的接收、存储、拉取等功能。

Producer&＃xff1a;生产者&＃xff0c;投递消息到Broker。

Consumer&＃xff1a;消费者&＃xff0c;从Broker获取消息&＃xff0c;处理业务逻辑。

Topic&＃xff1a;主题&＃xff0c;Broker收到某个主题的消息后&＃xff0c;会把消息推送给订阅了该主题的消费者。

JSON(Javascript Object Notation)&＃xff0c;JS对象简谱。是一种轻量级的数据交换格式&＃xff0c;采用完全独立于编程语言的文本格式来存储和表示数据。简洁和清晰的层次结构使得JSON成为理想的数据交换语言。易于人阅读和编写&＃xff0c;同时也易于机器解析和生成&＃xff0c;并有效地提升网络传输效率。

示例&＃xff1a;(在线json格式化校验&＃xff1a;https://www.bejson.com/)

{"group1": [{"name": "LiMing","gender": "male","age": "9"}, {"name": "Janny","gender": "female","age": "8","nationality": "Canada"}, {"name": "Danny","gender": "male","age": "10","species": "dinosaur"}],"group2": [{"name": "Kangkang","gender": "male","age": "14"}, {"name": "Michael","gender": "male","age": "14","nationality": "America"}, {"name": "Jane","gender": "female","age": "14","nationality": "Australia"}, {"name": "Maria","gender": "female","age": "15","nationality": "Cuba"}],"group3": {"name": "LiHua","gender": "female","age": "18","remarks": "eternalGod"}
}

1.4 MQ的简单实现

1.4.1 Maven依赖

com.alibabafastjson1.2.62

1.4.2 多线程实现本地MQ

思路&＃xff1a;

创建一个阻塞队列作为Broker&＃xff1b;创建两个线程&＃xff0c;一个作为Producer&＃xff0c;另一个作为Consumer。

Producer向Broker发送消息&＃xff0c;再由Consumer消费该消息。

代码演示&＃xff1a;

import com.alibaba.fastjson.JSONObject;import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;public class MyMQ {/*** Broker*/private static LinkedBlockingQueue broker &＃61; new LinkedBlockingQueue<>();public static void main(String[] args) {/*** Producer*/Thread producer &＃61; new Thread(() -> {while (true) {JSONObject data1 &＃61; new JSONObject();JSONObject data2 &＃61; new JSONObject();JSONObject data3 &＃61; new JSONObject();JSONObject data4 &＃61; new JSONObject();data1.put("name", "Kangkang");data1.put("gender", "male");data1.put("age", 14);data2.put("name", "Michael");data2.put("gender", "male");data2.put("age", 14);data3.put("name", "Jane");data3.put("gender", "female");data3.put("age", 14);data4.put("name", "Maria");data4.put("gender", "female");data4.put("age", 15);JSONObject data &＃61; new JSONObject();Object[] array &＃61; {data1, data2, data3, data4};data.put("group2", array);broker.offer(data);try {TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}}}, "生产者");producer.start();/*** Consumer*/Thread consumer &＃61; new Thread(() -> {while (true) {JSONObject data &＃61; broker.poll();if (data !&＃61; null) {System.out.println(Thread.currentThread().getName() &＃43; "获取到数据&＃xff1a;" &＃43; data.toJSONString());}}}, "消费者");consumer.start();}
}

运行结果&＃xff1a;(每5s输出一次)

消费者获取到数据&＃xff1a;{"group2":[{"gender":"male","name":"Kangkang","age":14},{"gender":"male","name":"Michael","age":14},{"gender":"female","name":"Jane","age":14},{"gender":"female","name":"Maria","age":15}]}

1.5 当前的主流MQ

1.5.1 ActiveMQ

ActiveMQ是Apache的产品&＃xff0c;是完全支持JMS(Java Message Service)规范的消息中间件&＃xff0c;提供了丰富的API&＃xff0c;支持各种各样的客户端去调用&＃xff0c;还提供了多种集群架构模式&＃xff0c;可以说是目前行业内最流行、能力最强劲的开源消息总线。

但是ActiveMQ在近几年的发展过程中&＃xff0c;它的性能一直受人诟病——它的吞吐量并不高&＃xff0c;所以大型的互联网公司一般都不会去用&＃xff0c;而在中小型企业中比较受欢迎。

1.5.2 Kafka

Kafka目前属于Apache顶级项目。Kafka追求高吞吐量&＃xff0c;一开始的目的就是用于日志的收集和传输。

虽然Kafka的性能特别高&＃xff0c;但它不支持事务&＃xff0c;对消息的重复、丢失、错误没有严格要求。

1.5.3 RocketMQ

RocketMQ是阿里开源的消息中间件&＃xff0c;它是纯Java开发&＃xff0c;具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka&＃xff0c;但并不是Kafka的一个Copy&＃xff0c;它对消息的可靠传输及事务性做了优化&＃xff0c;目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。

1.5.4 RabbitMQ

RabbitMQ是使用Erlang语言开发的开源消息队列&＃xff0c;基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由(包括点对点和发布/订阅)、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景&＃xff0c;对性能和吞吐量的要求还在其次。

【暂时写到这里&＃xff0c;如果以后有时间写写RabbitMQ和Kafka】

加油&＃xff01;(ง •_•)ง