RabbitMQ架构设计原理

消息中间件

消息中间件基于队列模型实现异步/同步传输数据

作用：可以实现支撑高并发、异步解耦、流量削峰、降低耦合度。

"异步" 和 "同步"

通常指的是数据发送（生产）和消费（处理）的方式。

解耦：

解耦意味着将原本紧密关联或相互依赖的组件、功能或系统分离，使它们能够独立地运行、修改和扩展，而不需要影响其他部分。

在进程或线程中，解耦可以使得多步骤的操作并行执行，而不是串行执行，从而提高效率。例如，在一个下载和解析的场景中，下载和解析函数可以放在进程池或线程池中并行运行，谁先下载完就先解析，从而实现解耦。

“异步解耦”

可以理解为在并发编程或系统设计中，通过异步处理的方式实现系统组件或功能之间的解耦，使得它们能够并行运行、独立扩展，并且不会相互阻塞或依赖。

流量削峰：

流量削峰是指通过一些技术手段来削弱瞬时的请求高峰，使系统吞吐量在高峰请求下保持可控

传统的http请求有诸多缺点：

1.在高并发的情况下，发送大量的请求达到服务器端导致服务器端处理请求堆积。

2.Tomcat服务器处理每个请求都有自己独立的线程，如果超过最大线程数，会将该请求缓存到队列中，请求堆积过多的情况下，可能导致tomcat服务器崩溃

所以一般都会在nginx入口实现限流，整合服务保护框架。

3. http请求处理业务逻辑比较耗时，容易造成客户端一直等待，阻塞等待过程中会导致客户端超时重发，引发幂等性问题。

注意事项：接口是为http协议的情况下，最好不要处理比较耗时的业务逻辑，耗时的业务逻辑应该单独交给多线程或者是mq处理。

举例说明：

客户端发送请求到达服务器端，服务器端实现会员注册业务逻辑，

1.insertMember() --插入会员数据 1s

2.sendSms()----发送登陆短信提醒 3s

3.sendCoupons()----发送新人优惠券 3s

总共响应需要7s时间，可能会导致客户端阻塞7s时间，对用户体验不是很好。

代码演示：

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class MemberService {

    //@Autowired
    //public MemberServiceAsync memberServiceAsync;

    //@Bean
    @RequestMapping("/addMember")
    public  String addMember(){
        //1.数据库插入数据 log.info(">01<");
        System.out.println(">01<");
        sms();
        System.out.println(">04<");
        return "用户注册成功！！";
    }


    public String sms(){
        System.out.println(">02<");
        try{
            System.out.println(">正在发送短信<");
            Thread.sleep(3000);
        }catch(Exception e){
            e.printStackTrace();
        }
        System.out.println(">03<");
        return "短信发送完成！";
    }
}

解决措施：

多线程与MQ方式实现异步

小项目可用多线程实现异步，大项目需要使用MQ，因为多线程面对高并发会对CPU造成损耗

1、多线程处理业务逻辑（实现异步操作）：

用户向数据库中插入一条数据之后，在单独开启一个线程异步发送短信和优惠操作。客户端只需要等待1s时间

优点：适合于小项目实现异步

缺点：有可能会消耗服务器cpu资源资源

代码演示：

异步类：

主配置类加注解：

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

//异步类实现sms方法
@Component
public class MemberServiceAsync
{
    @Async
    public String sms(){
        System.out.println(">02<");
        System.out.println(">正在发送短信<");
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(">03<");
        return "短信发送完成！！";
    }
}

解析：@Async的作用是什么

它用于声明一个方法是异步的。

当你在一个方法上使用了 @Async 注解，并正确配置了相关的异步支持，那么该方法将不会在主线程中同步执行，而是会由 Spring 的任务调度器（通常是一个线程池）异步地执行。

配置异步支持通过在配置类上添加 @EnableAsync 注解或在 XML 配置中添加来实现。

2、Mq处理业务逻辑（实现异步操作）：

当生产者一操作数据库存储数据，就会给消息中间件发送一个msg的提示消息。消息中间件的消息第一次发送给消费者时，会判断有没有消费者的存在，如果有，消费者会主动去消息队列里拉取消息。之后只要消费者存在，服务端（消息中间件）就属于将消息推送给消费端。

MQ实现的两个版本：

1、没有网络的情况下实现MQ ：利用多线程制造生产者和消费者

手撕代码：

package com.qcby.async;

import org.json.JSONObject;

import java.util.concurrent.LinkedBlockingDeque;

/**
 * 没有网络的情况下实现MQ
 * 利用多线程制造生产者和消费者
 */
public class BoyatopThreadMQ {

    //MQ服务器 初始化消息的队列
    private static LinkedBlockingDeque<JSONObject> msgs=new LinkedBlockingDeque();

    //主函数  程序入口
    public static void main(String[] args) {
        //生产者  生产线程
        Thread producerThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while(true){
                        Thread.sleep(1000);
                        JSONObject data = new JSONObject();
                        data.put("userId","12345");
                        //存入消息队列
                        msgs.offer(data);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },"生产者");
        producerThread.start();
        //消费端  消费线程
        Thread consumerThread = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while(true){
                        JSONObject data = msgs.poll();
                        if(data != null){
                            System.out.println(Thread.currentThread().getName() + ",获取到数据:" + data);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        },"消费者");
        consumerThread.start();
    }

}

2、基于网络通讯版本mq netty实现

解释：

消费者netty客户端与nettyServer端MQ服务器端保持长连接，MQ服务器端保存消费者连接。

生产者netty客户端发送请求给nettyServer端MQ服务器端，MQ服务器端在将该消息内容发送给消费者。

生产者投递消息给MQ服务器端，MQ服务器端需要缓存该消息。

如果mq服务器端宕机之后，消息如何保证不丢失？

答：持久化机制，会在磁盘中存储一份

如果mq接收到生产者投递消息，如果消费者不在的情况下，该消息是否会丢失？

答：不会丢失，因为有消息确认机制，必须要消费者消费该消息成功之后，再通知给mq服务器端删除该消息。

（mq与消费者之间的信息传递：先拉取，建立长连接，后推送）

对于多个消费者，如果没有主题（或者在一个主题里面）会不会消费同一条消息？

答：不会，每一个消费者消费完之后，会告诉MQ：消费完毕，可以删除，消息确认机制会避免重复性消费同组消费者不会出现重复性消费的情况，因为有消息确认机制

Mq如何实现抗高并发思想？

答：Mq消费者根据自身能力情况，拉取mq服务器端消息消费。默认的情况下是取出一条消息。

缺点：1、存在延迟的问题 2、需要考虑mq消费者提高速率的问题

如何消费者提高速率？

答：消费者实现集群、消费者批量获取消息即可。

Mq与多线程之间区别：

MQ可以实现异步/解耦/流量削峰问题；

多线程也可以实现异步，但是消耗到cpu资源，没有实现解耦。

Mq消息中间件名词

Producer 生产者：投递消息到MQ服务器端；

Consumer 消费者：从MQ服务器端获取消息处理业务逻辑；

Broker MQ服务器端：

Topic 主题：分类业务逻辑发送短信主题、发送优惠券主题

Queue 存放消息模型队列：先进先出后进后出原则数组/链表

Message 生产者投递消息报文：json 举例： body:{"msg":{"userId":"123456","age":"23"},"type":"producer"，”topic”:””}