五种服务异步通信（MQ）-详解、代码案例

server/2025/3/15 2:22:52/

简介：本篇文章主要是介绍了常用的异步通信原理，主要是RabbitMQ技术

目录

1、初始MQ（异步通讯）

1.1 同步通讯

1.2 异步通讯

1.3 MQ常见框架

2、RabbitMQ快速入门

2.1 RabbitMQ概述和安装

2.2 常见消息模型

2.3 快速入门

3、SpringAMQP

3.1 什么是SpringAMQP

3.2 SimpleQueue案例

3.3 SpringAMQP（发布、订阅模式）

3.3.1 广播模式

3.3.2 路由模式代码演示

3.3.3 话题模式

4、SpringAMQP-消息转换器

5、总结

1、初始MQ（异步通讯）

1.1 同步通讯

图 1.1-1 同步通讯存在的问题
上图中展示的就是同步通讯的问题

1.2 异步通讯

图 1.2-1 异步通讯优缺点

异步通信的优点：

  • 耦合度地
  • 吞吐量提升
  • 故障隔离
  • 流量削峰

异步通信的缺点：

  • 依赖于Broker的可靠性、安全性、吞吐能力
  • 架构复杂了、业务没有明显的流程线、不好追踪管理
上图中展示的就是异步通信的优缺点

1.3 MQ常见框架

图 1.3-1  MQ产品
上图中展示的便是四款常见的MQ产品，他们之间的优势性能也有清晰地比对

2、RabbitMQ快速入门

2.1 RabbitMQ概述和安装

图 2.1-1 RabbitMQ安装
所需要的安装包、详细记录安装步骤的MD文件，因为内容过多，我放在网盘里面了
百度网盘地址：https://pan.baidu.com/s/1FZtWCWMl_QpZEIcGNnpwKA 
提取码：6666
图 2.1-2 RabbitMQ概述
上图中展示的便是RabbitMQ的内部流程、逻辑，即消息发送者发送消息后传递给交换机，交换机将其消息存储到queue队列中，等待消息接受者获取

2.2 常见消息模型

图 2.2-1 五种消息模型
上图中展示的就是常用的五种消息队列模型，其官网地址：RabbitMQ Tutorials | RabbitMQ

2.3 快速入门

java">package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Testpublic void testSendMessage() throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.150.101");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.发送消息String message = "hello, rabbitmq!";channel.basicPublish("", queueName, null, message.getBytes());System.out.println("发送消息成功：【" + message + "】");// 5.关闭通道和连接channel.close();connection.close();}
}

java">package cn.itcast.mq.helloworld;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerTest {public static void main(String[] args) throws IOException, TimeoutException {// 1.建立连接ConnectionFactory factory = new ConnectionFactory();// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码factory.setHost("192.168.150.101");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("itcast");factory.setPassword("123321");// 1.2.建立连接Connection connection = factory.newConnection();// 2.创建通道ChannelChannel channel = connection.createChannel();// 3.创建队列String queueName = "simple.queue";channel.queueDeclare(queueName, false, false, false, null);// 4.订阅消息channel.basicConsume(queueName, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {// 5.处理消息String message = new String(body);System.out.println("接收到消息：【" + message + "】");}});System.out.println("等待接收消息。。。。");}
}

3、SpringAMQP

3.1 什么是SpringAMQP

图 3.1-1 SpringAMQP介绍
上图中展示的是关于SpringAMQP的消息发送和接收的标准

3.2 SimpleQueue案例

图 3.2-1 消息发送者
上图中展示的是消息发送者的代码案例：即配置连接信息、编写测试代码

3.3 SpringAMQP（发布、订阅模式）

图 3.3-1 发布、订阅模式
上图中展示的是三种通过路由器转发消息的模型，即广播模式、路由模式、话题模式
3.3.1 广播模式

1、消息发送者代码

java">package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testFanoutExchange(){// 交换机名称String exchangeName = "itcast.fanout";// 消息String message = "hello,everyone!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "", message);}
}

2、交换机、队列配置类代码

java">package cn.itcast.mq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class FanoutConfig {// 1.声明广播交换机@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("itcast.fanout");}@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}// 2.交换机绑定队列一@Beanpublic Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}// 3.交换机绑定队列二@Beanpublic Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}

3、消息接受者代码

java">package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.time.LocalTime;
@Component
public class SpringRabbitListener {/*    @RabbitListener(queues = "simple.queue")public void listenWorkQueue1(String msg) throws InterruptedException{System.out.println("消费者1接收到消息: 【" + msg + "】" + LocalTime.now());}*/@RabbitListener(queues = "fanout.queue1")public void listenFanoutQueue1(String msg){System.out.println("消费者1接收到消息: 【" + msg + "】" + LocalTime.now());}@RabbitListener(queues = "fanout.queue2")public void listenFanoutQueue2(String msg){System.out.println("消费者1接收到消息: 【" + msg + "】" + LocalTime.now());}
}
3.3.2 路由模式代码演示

1、消息发送者代码

java">package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testFanoutExchange(){// 交换机名称String exchangeName = "itcast.direct";// 消息String message = "hello,everyone!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "red", message);}
}

2、消息接受者代码

java">package cn.itcast.mq.listener;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalTime;@Component
public class SpringRabbitListener {@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"}))public void listenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"}))public void listenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}
}
3.3.3 话题模式

1、消息发送者代码

java">package cn.itcast.mq.helloworld;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class PublisherTest {@Autowiredprivate RabbitTemplate rabbitTemplate;
/*@Testpublic void testFanoutExchange(){// 交换机名称String exchangeName = "itcast.direct";// 消息String message = "hello,everyone!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "red", message);}*/@Testpublic void testTopicExchange(){// 交换机名称String exchangeName = "itcast.topic";// 消息String message = "hello,everyone!";// 发送消息rabbitTemplate.convertAndSend(exchangeName, "china.news", message);}
}

2、消息接受者代码

java">package cn.itcast.mq.listener;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.time.LocalTime;@Component
public class SpringRabbitListener {/*** 话题路由器*/@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue1"),exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),key = "china.#"))public void listenTopicQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "topic.queue2"),exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),key = "#.news"))public void listenTopicQueue2(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}
/*    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"), exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), key = {"red", "blue"}))public void listenDirectQueue1(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"), exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), key = {"red", "yellow"}))public void listenDirectQueue2(String msg){System.out.println("消费者接收到direct.queue1的消息: 【" + msg + "】");}*/
}

4、SpringAMQP-消息转换器

图 4-1 SpringAMQP的作用
我们知道 RabbitTemplate 传递的参数中，消息对象是以字节数组传递的，经过序列化（默认是通过JDK实现的）后显示为正常的数据，但是如果传递的是Map，List集合这种数据，SpringCloud自带的序列化就会出现异常，为了解决这一问题，我们需要引入SpringAMQP-消息转换器
图 4-1 项目的总pom文件
在项目的总pom文件中添加相对应的依赖
图 4-3 消息发送端、接收端
在项目的消息发送端、接收端的启动类中创建Bean对象

5、总结


http://www.ppmy.cn/server/9896.html

相关文章

大舍传媒国外活动策划，助您在国际舞台上大放异彩

大舍传媒国外活动策划，助您在国际舞台上大放异彩

一、引言 随着全球化的不断深入，越来越多的企业开始将目光投向国际市场。为了更好地拓展业务，提升企业品牌的国际影响力，各种海外活动策划和海外演议一站式服务需求日益增加。大舍传媒凭借多年的行业经验和专业团队，为您提供全方…
阅读更多...
SQL的基础语句

SQL的基础语句

1、select语句 select colums from table_name 2、条件语句 #查询出查询出用户id为1和3的用户记录 IN 操作符允许我们在 WHERE 子句中规定多个值。 select * from student where id in (1,3) #查询出所有姓王的同学 模糊查询 like 通配符(% 任意多个字符 _单个字符) #下例…
阅读更多...
OpenHarmony开发实例：【 待办事项TodoList】

OpenHarmony开发实例：【 待办事项TodoList】

简介 TodoList应用是基于OpenHarmony SDK开发的安装在润和HiSpark Taurus AI Camera(Hi3516d)开发板标准系统上的应用；应用主要功能是以列表的形式，展示需要完成的日程；通过本demo可以学习到 JS UI 框架List使用； 运行效果 样例…
阅读更多...
智慧安防边缘计算硬件AI智能分析网关V4算法启停的操作步骤

智慧安防边缘计算硬件AI智能分析网关V4算法启停的操作步骤

TSINGSEE青犀视频智能分析网关V4内置了近40种AI算法模型，支持对接入的视频图像进行人、车、物、行为等实时检测分析，上报识别结果，并能进行语音告警播放。硬件管理平台支持RTSP、GB28181协议、以及厂家私有协议接入，可兼容市面上常…
阅读更多...
vue3左树的全选和反选

vue3左树的全选和反选

<el-input v-model"filterText" placeholder"" style"width: 48%"/><el-button type"primary" click"handleSearch" class"ml-2">查找</el-button><el-radio-group v-model"form.choic…
阅读更多...
【SAP ME 12】SAP NWDS（eclipse）下载、安装，配置

【SAP ME 12】SAP NWDS（eclipse）下载、安装，配置

1、下载 1.1、描述 1.2、下载 2、安装 3、配置 3.1、域名映射
阅读更多...
查看项目go代码cpu利用率

查看项目go代码cpu利用率

1.代码添加&#xff1a; "net/http"_ "net/http/pprof"第二步&#xff0c;在代码开始运行的地方加上go func() {log.Println(http.ListenAndServe(":6060", nil))}() 2.服务器上防火墙把6060打开 3.电脑安装&#xff1a;Download | Graphviz …
阅读更多...
【C++提高】算法

【C++提高】算法

算法 一、遍历算法1. for_each2. transform 二、查找算法1. find2. find_if3. adjacent_find4. binary_search5. count6. count_if 三、排序算法1. sort2. random_shuffle3. merge4. reverse 四、拷贝和替换算法1. copy2. replace3. replace_if4. swap 五、算术生成算法1. accu…
阅读更多...
最新文章

Copyright @ 2022~2023