文章目录
- MQ用来做什么的
- MQ会有什么样的麻烦
- MQ消息队列模式分类
- MQ消息队列常用协议
- 市场主流四大MQ
- RabbitMQ项目开发
- RabbitMQ中的组成部分
MQ用来做什么的
省流 ：系统解耦、异步调用、流量削峰
系统解耦
首先举例下面这个场景，现有ABCDE五个系统，最初的时候BCD三个系统都要调用A系统的接口获取数据，一切都很正常，但是突然，D系统说：我不要了，你不用给我传数据了，A系统无奈，只能修改代码，将调用D系统的代码删除，这时候还没删除呢，E系统发送了请求，但是A系统这时候还没处理完D系统的请求，A系统卒！！！彻底崩溃。
上述场景中，BCDE都需要用到A系统提供的数据，A系统跟其他四个系统严重耦合，需要时时刻刻考虑其他四个系统要是挂了怎么办，需不需要重新发送数据给他们，这个时候的A系统内心是崩溃的。
但是如果使用了MQ之后 ，A系统的数据只需要放到MQ里面，其他的系统想请求获取数据只需要去MQ里面消费即可，如果突然不想请求了，就取消对MQ的消费就行了，A系统根本不需要考虑给谁去响应这个数据，也不需要去维护代码，也不用考虑其他系统是否调用成功，失败超时等情况。
异步调用
ABCD四个系统，A系统收到一个请求，需要在自己本地写库，还需要往BCD三个系统写库，A系统自己写本地库需要3ms，往其他系统写库相对较慢，B系统200ms ，C系统350ms，D系统400ms，这样算起来，整个功能从请求到响应的时间为3ms+200ms+350ms+400ms=953ms，接近一秒，对于用户来说，点个按钮要等这么长时间，基本是无法接受的，侧面也反映出这家研发人员技术不咋地。
一般的互联网企业，对于用户请求响应的时间要求在100ms-200ms之间，这样，用户的眼睛存在视觉暂停现象，用户响应时间在此范围内就可以了，所以上面的现象是不可取的。
如果用了MQ，用户发送请求到A系统耗时3ms，A系统发送三条消息到MQ，假如耗时5ms，用户从发送请求到相应3ms+5ms=8ms，仅用了8ms，用户的体验非常好。
流量削峰
JD系统每天0—19点，系统风平浪静，结果一到八点抢购的时候，每秒并发达到百万，假设JD数据库没秒能处理1.5w条并发请求（并非实际数据，主要为了举例）,到八点抢购的时候，每秒并发百万，这直接导致系统异常，但是八点一过，可能也就几万用户在线操作，每秒的请求可能也就几百条，对整个系统毫无压力。
如果使用了MQ，每秒百万个请求写入MQ，因为JD系统每秒能处理1W+的请求，JD系统处理完然后再去MQ里面，再拉取1W+的请求处理，每次不要超过自己能处理的最大请求量就ok，这样下来，等到八点高峰期的时候，系统也不会挂掉，但是近一个小时内，系统处理请求的速度是肯定赶不上用户的并发请求的，所以都会积压在MQ中，甚至可能积压千万条，但是高峰期过后，每秒只会有一千多的并发请求进入MQ，但是JD系统还是会以每秒1W+的速度处理请求，所以高峰期一过，JD系统会很快消化掉积压在MQ的请求，在用户那边可能也就是等的时间长一点，但是绝对不会让系统挂掉。
MQ会有什么样的麻烦
系统可用性降低：
系统引入的外部依赖越多，系统要面对的风险越高，拿场景一来说，本来ABCD四个系统配合的好好的，没啥问题，但是你偏要弄个MQ进来插一脚，虽然好处挺多，但是万一MQ挂掉了呢，那样你系统不也就挂掉了。
系统复杂程度提高
非要加个MQ进来，如何保证没有重复消费呢？如何处理消息丢失的情况？怎么保证消息传递的顺序？问题太多。
一致性的问题
A系统处理完再传递给MQ就直接返回成功了，用户以为你这个请求成功了，但是，如果在BCD的系统里，BC两个系统写库成功，D系统写库失败了怎么办，这样就导致数据不一致了。
所以。消息队列其实是一套非常复杂的架构，你在享受MQ带来的好处的同时，也要做各种技术方案把MQ带来的一系列的问题解决掉，等一切都做好之后，系统的复杂程度硬生生提高了一个等级。
MQ消息队列模式分类
1. 点对点PTP
消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息。 消息被消费以后，queue中不再存储，所以消息消费者不可能消费到已经被消费的消息。 Queue支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
2. 发布/订阅
topic实现了发布和订阅，当你发布一个消息，所有订阅这个topic的服务都能得到这个消息，所以从1到N个订阅者都能得到一个消息的拷贝。
MQ消息队列常用协议
- AMQP协议
- MQTT协议
- STOMP协议
- XMPP协议
市场主流四大MQ
kafka、ActiveMQ、RabbitMQ、RocketMQ
RabbitMQ项目开发
超详细RabbitMQ入门，这一篇就够了
1. 安装erLang语言，配置环境变量
管网安装下载Winsdow10版本的erlang安装包
下载之后就得到这个东西
双击安装，一直点next就行了，安装完之后，配置环境变量
使用cmd命令，输入 erl -version 验证：
2. 安装RabbitMQ服务器
在RabbitMQ的gitHub项目中，下载windows版本的服务端安装包
下载之后得到这个东西
接着到双击安装，一直点下一步安装即可，安装完成后，找到安装目录
在此目录下打开cmd命令，进入到sbin目录下，输入rabbitmq-plugins enable rabbitmq_management命令安装管理页面的插件：
然后双击rabbitmq-server.bat启动脚本，然后打开服务管理可以看到RabbitMQ正在运行：
这时，打开浏览器输入http://localhost:15672，账号密码默认是：guest/guest
到这一步，安装就大功告成了！
3. 开始HelloWorld
导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
在application.yml文件写入配置信息
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
配置公共项目common
public class RabbitConfig{
// RabbitMQ的队列主题名称
public static final String RABBITMQ_DEMO_TOPIC = "rabbitmqDemoTopic";
//RabbitMQ的DIRECT交换机名称
public static final String RABBITMQ_DEMO_DIRECT_EXCHANGE = "rabbitmqDemoDirectExchange";
//RabbitMQ的DIRECT交换机和队列绑定的匹配键DirectRouting
public static final String RABBITMQ_DEMO_DIRECT_ROUTING = "rabbitmqDemoDirectRouting";
}
RabbitMQ中间组件的配置
@Configuration
public class DirectRabbitConfig {
@Bean
public Queue rabbitmqDemoDirectQueue() {
/**
* 1、name: 队列名称
* 2、durable: 是否持久化
* 3、exclusive: 是否独享、排外的。如果设置为true，定义为排他队列。则只有创建者可以使用此队列。也就是private私有的。
* 4、autoDelete: 是否自动删除。也就是临时队列。当最后一个消费者断开连接后，会自动删除。
* */
return new Queue(RabbitMQConfig.RABBITMQ_DEMO_TOPIC, true, false, false);
}
@Bean
public DirectExchange rabbitmqDemoDirectExchange() {
//Direct交换机
return new DirectExchange(RabbitMQConfig.RABBITMQ_DEMO_DIRECT_EXCHANGE, true, false);
}
@Bean
public Binding bindDirect() {
//链式写法，绑定交换机和队列，并设置匹配键
return BindingBuilder
//绑定队列
.bind(rabbitmqDemoDirectQueue())
//到交换机
.to(rabbitmqDemoDirectExchange())
//并设置匹配键
.with(RabbitMQConfig.RABBITMQ_DEMO_DIRECT_ROUTING);
}
}
生产者
@Service
public class RabbitMQServiceImpl implements RabbitMQService {
//日期格式化
private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
@Resource
private RabbitTemplate rabbitTemplate;
@Override
public String sendMsg(String msg) throws Exception {
try {
String msgId = UUID.randomUUID().toString().replace("-", "").substring(0, 32);
String sendTime = sdf.format(new Date());
Map<String, Object> map = new HashMap<>();
map.put("msgId", msgId);
map.put("sendTime", sendTime);
map.put("msg", msg);
rabbitTemplate.convertAndSend(RabbitMQConfig.RABBITMQ_DEMO_DIRECT_EXCHANGE, RabbitMQConfig.RABBITMQ_DEMO_DIRECT_ROUTING, map);
return "ok";
} catch (Exception e) {
e.printStackTrace();
return "error";
}
}
}
@RestController
@RequestMapping("/mall/rabbitmq")
public class RabbitMQController {
@Resource
private RabbitMQService rabbitMQService;
/**
* 发送消息
* @author java技术爱好者
*/
@PostMapping("/sendMsg")
public String sendMsg(@RequestParam(name = "msg") String msg) throws Exception {
return rabbitMQService.sendMsg(msg);
}
}
消费者
@Component
@RabbitListener(queues = {RabbitMQConfig.RABBITMQ_DEMO_TOPIC})
public class RabbitDemoConsumer{
@RabbitHandler
public void process(Map map){
System.out.println("消费者RabbitDemoConsumer从RabbitMQ服务端消费消息："+map.toString());
}
}
开始运行
运行前注意先运行服务器创建队列，再运行客户端获取消息。如果顺序相反，客户端先执行就会报错。
现在开始先启动生产者，发送一条消息，
最后再启动消费者，进行消费。
避免没有队列
由于队列不存在，启动消费者报错的这个问题。最好的方法是生产者和消费者都尝试创建队列，怎么写呢，有很多方式，我这里用一个相对简单一点的：
//实现BeanPostProcessor类，使用Bean的生命周期函数
@Component
public class DirectRabbitConfig implements BeanPostProcessor {
//这是创建交换机和队列用的rabbitAdmin对象
@Resource
private RabbitAdmin rabbitAdmin;
//初始化rabbitAdmin对象
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
// 只有设置为 true，spring 才会加载 RabbitAdmin 这个类
rabbitAdmin.setAutoStartup(true);
return rabbitAdmin;
}
//实例化bean后，也就是Bean的后置处理器
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
//创建交换机
rabbitAdmin.declareExchange(rabbitmqDemoDirectExchange());
//创建队列
rabbitAdmin.declareQueue(rabbitmqDemoDirectQueue());
return null;
}
}
这样启动生产者就会自动创建交换机和队列，不用等到发送消息才创建。
消费者需要加一点代码：
@Component
//使用queuesToDeclare属性，如果不存在则会创建队列
@RabbitListener(queuesToDeclare = @Queue(RabbitMQConfig.RABBITMQ_DEMO_TOPIC))
public class RabbitDemoConsumer {
//...省略
}
这样，无论生产者还是消费者先启动都不会出现问题了~
RabbitMQ中的组成部分
Exchange：消息队列交换机。按一定的规则将消息路由转发到某个队列。
Queue：消息队列，存储消息的队列。
Producer：消息生产者。生产方客户端将消息同交换机路由发送到队列中。
Consumer：消息消费者。消费队列中存储的消息。
