柚子快報(bào)激活碼778899分享：分布式 RabbitMQ知識(shí)點(diǎn)

http://yzkb.51969.com/

MQ架構(gòu)設(shè)計(jì)原理

什么是消息中間件

消息中間件基于隊(duì)列模型實(shí)現(xiàn)異步/同步傳輸數(shù)據(jù)

作用：可以實(shí)現(xiàn)支撐高并發(fā)、異步解耦、流量削峰、降低耦合度。

傳統(tǒng)的http請(qǐng)求存在那些缺點(diǎn)

1.Http請(qǐng)求基于請(qǐng)求與響應(yīng)的模型，在高并發(fā)的情況下，客戶端發(fā)送大量的請(qǐng)求達(dá)到

服務(wù)器端有可能會(huì)導(dǎo)致我們服務(wù)器端處理請(qǐng)求堆積。

2.Tomcat服務(wù)器處理每個(gè)請(qǐng)求都有自己獨(dú)立的線程，如果超過(guò)最大線程數(shù)會(huì)將該請(qǐng)求緩存到隊(duì)列中，如果請(qǐng)求堆積過(guò)多的情況下，有可能會(huì)導(dǎo)致tomcat服務(wù)器崩潰的問(wèn)題。

所以一般都會(huì)在nginx入口實(shí)現(xiàn)限流，整合服務(wù)保護(hù)框架。

?3.http請(qǐng)求處理業(yè)務(wù)邏輯如果比較耗時(shí)的情況下，容易造成客戶端一直等待，阻塞等待過(guò)程中會(huì)導(dǎo)致客戶端超時(shí)發(fā)生重試策略，有可能會(huì)引發(fā)冪等性問(wèn)題。

注意事項(xiàng)：接口是為http協(xié)議的情況下，最好不要處理比較耗時(shí)的業(yè)務(wù)邏輯，耗時(shí)的業(yè)務(wù)邏輯應(yīng)該單獨(dú)交給多線程或者是mq處理。

Mq應(yīng)用場(chǎng)景有那些 消息隊(duì)列（Message Queue，簡(jiǎn)稱MQ）在分布式系統(tǒng)中有著廣泛的應(yīng)用，以下是MQ的一些主要應(yīng)用場(chǎng)景： 1.異步處理： - 當(dāng)某個(gè)操作不需要立即返回結(jié)果，或者該操作非常耗時(shí)且影響主業(yè)務(wù)流程的性能時(shí)，可以使用MQ將操作放入后臺(tái)異步執(zhí)行。例如，用戶注冊(cè)成功后發(fā)送歡迎郵件或短信，這些操作不需要實(shí)時(shí)同步完成，可以放入MQ中異步處理。 2.應(yīng)用解耦： - MQ允許應(yīng)用間進(jìn)行低耦合的通信。通過(guò)將消息發(fā)送到一個(gè)或多個(gè)目的地址（隊(duì)列或主題），應(yīng)用程序可以獨(dú)立地執(zhí)行和擴(kuò)展，而不需要知道彼此的實(shí)現(xiàn)細(xì)節(jié)。例如，訂單系統(tǒng)產(chǎn)生訂單后，將訂單數(shù)據(jù)發(fā)送到MQ，庫(kù)存系統(tǒng)、物流系統(tǒng)等多個(gè)系統(tǒng)監(jiān)聽(tīng)MQ中的訂單數(shù)據(jù)，根據(jù)訂單數(shù)據(jù)執(zhí)行相應(yīng)的業(yè)務(wù)邏輯。 3.流量削峰： - 在高并發(fā)場(chǎng)景下，如果某個(gè)應(yīng)用或系統(tǒng)突然接收到大量的請(qǐng)求，可能導(dǎo)致系統(tǒng)壓力過(guò)大甚至崩潰。通過(guò)使用MQ，可以將這些請(qǐng)求暫存起來(lái)，然后按照系統(tǒng)能夠處理的速率勻速地將請(qǐng)求分發(fā)給后臺(tái)服務(wù)進(jìn)行處理。這樣可以有效地緩解系統(tǒng)壓力，避免系統(tǒng)崩潰。 4.數(shù)據(jù)同步： - MQ可以實(shí)現(xiàn)不同系統(tǒng)之間的數(shù)據(jù)同步。例如，在分布式數(shù)據(jù)庫(kù)中，可以使用MQ將數(shù)據(jù)的變更事件（如增刪改操作）發(fā)布到MQ中，其他系統(tǒng)監(jiān)聽(tīng)這些事件并據(jù)此更新自己的數(shù)據(jù)副本。 5.日志處理： - MQ可以用于收集、處理和存儲(chǔ)日志信息。通過(guò)將日志信息發(fā)送到MQ中，可以異步地將日志信息寫(xiě)入到存儲(chǔ)系統(tǒng)（如HDFS、Elasticsearch等）中，并進(jìn)行后續(xù)的分析和處理。

-6.任務(wù)調(diào)度： - MQ可以用于實(shí)現(xiàn)任務(wù)的調(diào)度和執(zhí)行。通過(guò)將任務(wù)信息發(fā)送到MQ中，并指定任務(wù)的執(zhí)行時(shí)間和執(zhí)行條件，可以實(shí)現(xiàn)定時(shí)任務(wù)、延遲任務(wù)等功能的實(shí)現(xiàn)。 7.微服務(wù)架構(gòu)中的通信： - 在微服務(wù)架構(gòu)中，服務(wù)與服務(wù)之間的通信是常見(jiàn)的需求。使用MQ可以實(shí)現(xiàn)服務(wù)之間的松耦合通信，使得服務(wù)可以獨(dú)立地?cái)U(kuò)展和升級(jí)。同時(shí)，MQ還可以提供消息持久化、消息確認(rèn)等機(jī)制，保證服務(wù)之間通信的可靠性和穩(wěn)定性。 8.分布式事務(wù)： - 在分布式系統(tǒng)中，實(shí)現(xiàn)跨多個(gè)服務(wù)的事務(wù)一致性是一個(gè)挑戰(zhàn)。通過(guò)使用MQ的分布式事務(wù)功能（如RabbitMQ的RabbitMQ-delayed-message-exchange插件或Kafka的事務(wù)API），可以實(shí)現(xiàn)跨多個(gè)服務(wù)的分布式事務(wù)處理。 9.消息廣播與訂閱： - MQ可以實(shí)現(xiàn)消息的廣播與訂閱功能。發(fā)布者將消息發(fā)送到特定的主題（Topic），所有訂閱了該主題的消費(fèi)者都可以接收到這些消息。這種模式常用于實(shí)時(shí)消息系統(tǒng)、新聞推送等場(chǎng)景。 10.跨語(yǔ)言通信： - MQ支持多種編程語(yǔ)言和平臺(tái)，可以實(shí)現(xiàn)跨語(yǔ)言、跨平臺(tái)的通信。這使得不同語(yǔ)言開(kāi)發(fā)的應(yīng)用或服務(wù)可以方便地進(jìn)行通信和數(shù)據(jù)交換。

為什么使用MQ

消息中間件（如RabbitMQ、Kafka、ActiveMQ等）在分布式系統(tǒng)中扮演著重要的角色，它們基于隊(duì)列模型（Queue Model）實(shí)現(xiàn)數(shù)據(jù)的異步或同步傳輸。下面我將詳細(xì)解釋這些中間件如何支持高并發(fā)、異步解耦、流量削峰以及降低耦合度的作用。 1.支撐高并發(fā)： - 當(dāng)系統(tǒng)面臨高并發(fā)請(qǐng)求時(shí)，如果每個(gè)請(qǐng)求都直接由系統(tǒng)處理，可能會(huì)導(dǎo)致系統(tǒng)資源迅速耗盡，響應(yīng)速度下降，甚至服務(wù)崩潰。 - 消息中間件通過(guò)隊(duì)列的方式將請(qǐng)求暫存起來(lái)，后臺(tái)服務(wù)可以異步地從隊(duì)列中獲取請(qǐng)求并進(jìn)行處理。這樣，即使在高并發(fā)場(chǎng)景下，系統(tǒng)也能保持穩(wěn)定的響應(yīng)速度，因?yàn)檎?qǐng)求被均勻地分散到不同的時(shí)間段內(nèi)處理。 - 消息中間件通常支持多消費(fèi)者模型，即多個(gè)服務(wù)實(shí)例可以同時(shí)從隊(duì)列中消費(fèi)消息，從而進(jìn)一步提高并發(fā)處理能力。 2.異步解耦： - 在傳統(tǒng)的同步請(qǐng)求-響應(yīng)模型中，請(qǐng)求方必須等待響應(yīng)方處理完請(qǐng)求后才能繼續(xù)執(zhí)行后續(xù)操作。這種模型下，請(qǐng)求方和響應(yīng)方之間存在緊密的耦合關(guān)系。 - 消息中間件引入了異步通信機(jī)制，使得請(qǐng)求方和響應(yīng)方可以解耦。請(qǐng)求方只需將請(qǐng)求發(fā)送到消息隊(duì)列中，然后立即返回，無(wú)需等待響應(yīng)。響應(yīng)方可以在后臺(tái)異步地處理這些請(qǐng)求，并通過(guò)其他方式（如回調(diào)函數(shù)、HTTP響應(yīng)等）將結(jié)果通知給請(qǐng)求方。 - 這種異步解耦的方式可以提高系統(tǒng)的響應(yīng)速度和吞吐量，同時(shí)減少系統(tǒng)間的依賴關(guān)系，提高系統(tǒng)的可擴(kuò)展性和可維護(hù)性。 3.流量削峰： - 在某些場(chǎng)景下，系統(tǒng)可能會(huì)在短時(shí)間內(nèi)接收到大量的請(qǐng)求，導(dǎo)致系統(tǒng)負(fù)載急劇上升，甚至超過(guò)系統(tǒng)的處理能力。 - 消息中間件可以將這些請(qǐng)求暫存到隊(duì)列中，然后按照系統(tǒng)能夠處理的速率勻速地將請(qǐng)求分發(fā)給后臺(tái)服務(wù)進(jìn)行處理。這樣，即使系統(tǒng)面臨巨大的流量沖擊，也能保持穩(wěn)定的性能和響應(yīng)速度。 - 流量削峰不僅可以保護(hù)系統(tǒng)免受流量沖擊的影響，還可以提高系統(tǒng)的可用性和容錯(cuò)能力。 4.降低耦合度： - 在復(fù)雜的分布式系統(tǒng)中，各個(gè)服務(wù)之間可能存在復(fù)雜的依賴關(guān)系，這些依賴關(guān)系可能導(dǎo)致系統(tǒng)難以維護(hù)和擴(kuò)展。 - 消息中間件通過(guò)引入消息隊(duì)列作為服務(wù)之間的通信媒介，降低了服務(wù)之間的耦合度。服務(wù)之間不再需要直接調(diào)用對(duì)方的方法或接口，而是通過(guò)發(fā)送和接收消息來(lái)進(jìn)行通信。 - 這種基于消息的通信方式使得服務(wù)可以更加獨(dú)立地設(shè)計(jì)和實(shí)現(xiàn)，提高了系統(tǒng)的可維護(hù)性和可擴(kuò)展性。同時(shí)，由于服務(wù)之間的耦合度降低，系統(tǒng)的容錯(cuò)能力也得到了提高，即使某個(gè)服務(wù)出現(xiàn)故障，也不會(huì)影響其他服務(wù)的正常運(yùn)行。

Mq與多線程之間區(qū)別

MQ可以實(shí)現(xiàn)異步/解耦/流量削峰問(wèn)題；

多線程也可以實(shí)現(xiàn)異步，但是消耗到cpu資源，沒(méi)有實(shí)現(xiàn)解耦。

實(shí)例：（基于springboot實(shí)現(xiàn)異步多線程）

package com.example.rabbitmq.mq;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.scheduling.annotation.Async;

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.數(shù)據(jù)庫(kù)插入數(shù)據(jù) log.info(">01<");

System.out.println(">01<");

// sms();

memberServiceAsync.smsAsync();

System.out.println(">04<");

return "用戶注冊(cè)成功??！";

}

public String sms(){

System.out.println(">02<");

try{

System.out.println(">正在發(fā)送短信<");

Thread.sleep(3000);

}catch(Exception e){

e.printStackTrace();

}

System.out.println(">03<");

return "短信發(fā)送完成！";

}

}

package com.example.rabbitmq.mq;

import org.springframework.scheduling.annotation.Async;

import org.springframework.stereotype.Component;

@Component

public class MemberServiceAsync {

@Async

public void smsAsync(){ // 注意返回類型為void，因?yàn)楫惒椒椒ㄍǔ２环祷亟Y(jié)果給調(diào)用者

System.out.println(">02<");

try{

System.out.println(">正在發(fā)送短信<");

Thread.sleep(3000);

}catch(Exception e){

e.printStackTrace();

}

System.out.println(">03<");

// 注意：這里不應(yīng)該返回任何值給調(diào)用者，因?yàn)檎{(diào)用是異步的

}

}

啟動(dòng)類

package com.example.rabbitmq;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.Bean;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@EnableAsync

@SpringBootApplication

public class RabbitMqApplication {

public static void main(String[] args) {

SpringApplication.run(RabbitMqApplication.class, args);

}

}

效果

基于多線程隊(duì)列簡(jiǎn)單實(shí)現(xiàn)mq

package com.example.rabbitmq.mq;

import org.json.JSONObject;

import java.util.concurrent.LinkedBlockingDeque;

/**

* 沒(méi)有網(wǎng)絡(luò)的情況下實(shí)現(xiàn)MQ

* 利用多線程制造生產(chǎn)者和消費(fèi)者

*/

public class BoyatopThreadMQ {

//MQ服務(wù)器 初始化消息的隊(duì)列

private static LinkedBlockingDeque msgs=new LinkedBlockingDeque();

//主函數(shù) 程序入口

public static void main(String[] args) {

//生產(chǎn)者 生產(chǎn)線程

Thread producerThread = new Thread(new Runnable() {

@Override

public void run() {

try {

while(true){

Thread.sleep(1000);

JSONObject data = new JSONObject();

data.put("userId","12345");

//存入消息隊(duì)列

msgs.offer(data);

}

} catch (Exception e) {

e.printStackTrace();

}

}

},"生產(chǎn)者");

producerThread.start();

//消費(fèi)端 消費(fèi)線程

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() + ",獲取到數(shù)據(jù):" + data);

}

}

} catch (Exception e) {

e.printStackTrace();

}

}

},"消費(fèi)者");

consumerThread.start();

}

}

效果：

基于netty實(shí)現(xiàn)mq

Netty是一個(gè)高性能、異步事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架，專門用于構(gòu)建高性能、高可靠性的網(wǎng)絡(luò)應(yīng)用程序。它提供了一個(gè)快速簡(jiǎn)便的方法來(lái)構(gòu)建網(wǎng)絡(luò)服務(wù)器和客戶端，并且不需要關(guān)注復(fù)雜的網(wǎng)絡(luò)編程細(xì)節(jié)。Netty支持多種網(wǎng)絡(luò)協(xié)議，包括TCP、UDP、HTTP、WebSocket等，使其適用于各種應(yīng)用場(chǎng)景。Netty在許多大型互聯(lián)網(wǎng)公司和開(kāi)源項(xiàng)目中被廣泛使用，特別是在需要處理大量并發(fā)連接和低延遲的場(chǎng)景中。

消費(fèi)者netty客戶端與nettyServer端MQ服務(wù)器端保持長(zhǎng)連接，MQ服務(wù)器端保存消費(fèi)者連接。

生產(chǎn)者netty客戶端發(fā)送請(qǐng)求給nettyServer端MQ服務(wù)器端，MQ服務(wù)器端在將該消息內(nèi)容發(fā)送給消費(fèi)者

導(dǎo)入相關(guān)依賴

com.alibaba

fastjson

1.2.62

io.netty

netty-all

4.0.23.Final

com.alibaba

fastjson

1.2.62

org.apache.commons

commons-lang3

3.11

com.rabbitmq

amqp-client

3.3.4

Netty生產(chǎn)端

package com.example.rabbitmq.netty;

import com.alibaba.fastjson.JSONObject;

import io.netty.bootstrap.Bootstrap;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.*;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioSocketChannel;

/**

* @ClassName BoyatopNettyMQProducer

* @Author

* @Version V1.0

**/

public class NettyProducer {

public void connect(int port, String host) throws Exception {

//配置客戶端NIO 線程組

EventLoopGroup group = new NioEventLoopGroup();

Bootstrap client = new Bootstrap();

try {

client.group(group)

// 設(shè)置為Netty客戶端

.channel(NioSocketChannel.class)

/**

* ChannelOption.TCP_NODELAY參數(shù)對(duì)應(yīng)于套接字選項(xiàng)中的TCP_NODELAY,該參數(shù)的使用與Nagle算法有關(guān)。

* Nagle算法是將小的數(shù)據(jù)包組裝為更大的幀然后進(jìn)行發(fā)送，而不是輸入一次發(fā)送一次，因此在數(shù)據(jù)包不足的時(shí)候會(huì)等待其他數(shù)據(jù)的到來(lái)，組裝成大的數(shù)據(jù)包進(jìn)行發(fā)送，雖然該算法有效提高了網(wǎng)絡(luò)的有效負(fù)載，但是卻造成了延時(shí)。

* 而該參數(shù)的作用就是禁止使用Nagle算法，使用于小數(shù)據(jù)即時(shí)傳輸。和TCP_NODELAY相對(duì)應(yīng)的是TCP_CORK，該選項(xiàng)是需要等到發(fā)送的數(shù)據(jù)量最大的時(shí)候，一次性發(fā)送數(shù)據(jù)，適用于文件傳輸。

*/

.option(ChannelOption.TCP_NODELAY, true)

.handler(new ChannelInitializer() {

@Override

protected void initChannel(SocketChannel ch) throws Exception {

ch.pipeline().addLast(new NettyProducer.NettyClientHandler());

1. 演示LineBasedFrameDecoder編碼器

// ch.pipeline().addLast(new LineBasedFrameDecoder(1024));

// ch.pipeline().addLast(new StringDecoder());

}

});

//綁定端口, 異步連接操作

ChannelFuture future = client.connect(host, port).sync();

//等待客戶端連接端口關(guān)閉

future.channel().closeFuture().sync();

} finally {

//優(yōu)雅關(guān)閉 線程組

group.shutdownGracefully();

}

}

public static void main(String[] args) {

int port = 9008;

NettyProducer client = new NettyProducer();

try {

client.connect(port, "127.0.0.1");

} catch (Exception e) {

e.printStackTrace();

}

}

public class NettyClientHandler extends ChannelInboundHandlerAdapter {

@Override

public void channelActive(ChannelHandlerContext ctx) throws Exception {

JSONObject data = new JSONObject();

data.put("type", "producer");

JSONObject msg = new JSONObject();

msg.put("userId", "123456");

msg.put("age", "23");

data.put("msg", msg);

// 生產(chǎn)發(fā)送數(shù)據(jù)

byte[] req = data.toJSONString().getBytes();

ByteBuf firstMSG = Unpooled.buffer(req.length);

firstMSG.writeBytes(req);

ctx.writeAndFlush(firstMSG);

}

/**

* 客戶端讀取到服務(wù)器端數(shù)據(jù)

*

* @param ctx

* @param msg

* @throws Exception

*/

@Override

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

ByteBuf buf = (ByteBuf) msg;

byte[] req = new byte[buf.readableBytes()];

buf.readBytes(req);

String body = new String(req, "UTF-8");

System.out.println("客戶端接收到服務(wù)器端請(qǐng)求:" + body);

}

// tcp屬于雙向傳輸

@Override

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

ctx.close();

}

}

}

Netty服務(wù)端

package com.example.rabbitmq.netty;

import com.alibaba.fastjson.JSONObject;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.buffer.ByteBuf;

import io.netty.buffer.Unpooled;

import io.netty.channel.*;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import org.apache.commons.lang3.StringUtils;

import java.io.UnsupportedEncodingException;

import java.util.ArrayList;

import java.util.concurrent.LinkedBlockingDeque;

/**

* @ClassName

* @Author

* @Version V1.0

**/

public class NettyMQServer {

public void bind(int port) throws Exception {

/**

* Netty 抽象出兩組線程池BossGroup和WorkerGroup

* BossGroup專門負(fù)責(zé)接收客戶端的連接, WorkerGroup專門負(fù)責(zé)網(wǎng)絡(luò)的讀寫(xiě)。

*/

EventLoopGroup bossGroup = new NioEventLoopGroup();

EventLoopGroup workerGroup = new NioEventLoopGroup();

ServerBootstrap bootstrap = new ServerBootstrap();

try {

bootstrap.group(bossGroup, workerGroup)

// 設(shè)定NioServerSocketChannel 為服務(wù)器端

.channel(NioServerSocketChannel.class)

//BACKLOG用于構(gòu)造服務(wù)端套接字ServerSocket對(duì)象，標(biāo)識(shí)當(dāng)服務(wù)器請(qǐng)求處理線程全滿時(shí)，

//用于臨時(shí)存放已完成三次握手的請(qǐng)求的隊(duì)列的最大長(zhǎng)度。如果未設(shè)置或所設(shè)置的值小于1，Java將使用默認(rèn)值50。

.option(ChannelOption.SO_BACKLOG, 100)

// 服務(wù)器端監(jiān)聽(tīng)數(shù)據(jù)回調(diào)Handler

.childHandler(new NettyMQServer.ChildChannelHandler());

//綁定端口, 同步等待成功;

ChannelFuture future = bootstrap.bind(port).sync();

System.out.println("當(dāng)前服務(wù)器端啟動(dòng)成功...");

//等待服務(wù)端監(jiān)聽(tīng)端口關(guān)閉

future.channel().closeFuture().sync();

} catch (Exception e) {

e.printStackTrace();

} finally {

//優(yōu)雅關(guān)閉 線程組

bossGroup.shutdownGracefully();

workerGroup.shutdownGracefully();

}

}

private class ChildChannelHandler extends ChannelInitializer {

@Override

protected void initChannel(SocketChannel ch) throws Exception {

// 設(shè)置異步回調(diào)監(jiān)聽(tīng)

ch.pipeline().addLast(new NettyMQServer.MayiktServerHandler());

}

}

public static void main(String[] args) throws Exception {

int port = 9008;

new NettyMQServer().bind(port);

}

private static final String type_consumer = "consumer";

private static final String type_producer = "producer";

private static LinkedBlockingDeque msgs = new LinkedBlockingDeque<>();

private static ArrayList ctxs = new ArrayList<>();

// 生產(chǎn)者投遞消息的：topicName

public class MayiktServerHandler extends SimpleChannelInboundHandler