RabbitMQ从入门到实战（图文并茂）

MQ概述

MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信。

MQ优势

1、应用解耦

MQ相当于一个中介，生产方通过MQ与消费方交互，它将应用程序进行解耦合。

2、任务异步处理

3、削峰填谷

如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。

消息被MQ保存起来了，然后系统就可以按照自己的消费能力来消费，比如每秒1000个消息，这样慢慢写入数据库，这样就不会卡死数据库了。

但是使用了MQ之后，限制消费消息的速度为1000，但是这样一来，高峰期产生的数据势必会被积压在 MQ中，高峰就被“削”掉了。但是因为消息积压，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000QPS，直到消费完积压的消息,这就叫做“填谷”

MQ的劣势

系统可用性降低系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高 MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题 A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？

实现MQ的大致有两种主流方式：AMQP、JMS。

AMQP 与 JMS 区别

`AMQP`	`JMS`
通过规定协议来统一数据交互的格式	定义了统一的接口，来对消息进行统一
只是协议，不规定实现方式，因此是跨语言的	限定了必须使用`java`语言
消息模式单一	规定了两种消息模式

RabbitMQ官方地址：Messaging that just works — RabbitMQ

2007年，Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言专门为开发高并发和分布式系统的一种语言，在电信领域使用广泛。 RabbitMQ 基础架构如下图

官网对应模式介绍：RabbitMQ Tutorials — RabbitMQ

添加依赖

Work Queues 与入门程序的简单模式的代码是几乎一样的；可以完全复制，并复制多一个消费者进行多个消费者同时消费消息的测试。

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到

发布订阅模式与工作队列模式的区别

1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息（底层使用默认交换机）。

3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机。

在编码上与 Publish/Subscribe发布与订阅模式的区别是交换机的类型为：Direct，还有队列绑定交换机的时候需要指定routing key

Routing模式要求队列在绑定交换机时要指定routing key，消息会转发到符合routing key的队列。

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式和 Routing路由模式的功能；只是Topic 在配置routing key 的时候可以使用通配符，显得更加灵活

模式总结

RabbitMQ工作模式：

简单模式 `HELLOWORLD`	一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）
工作队列模式 `Work Queue`	一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）
发布订阅模式 `Publish/subscribe`	需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列
路由模式 Routing	需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列
通配符模式 Topic	需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的 routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

RabbitMQ整合Spring

创建工程模块，添加依赖

<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0"         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion><groupId>com.lxs</groupId>
<artifactId>spring-rabbitmq-producer</artifactId>
<version>1.0-SNAPSHOT</version><dependencies>        <dependency>            <groupId>org.springframework</groupId>            <artifactId>spring-context</artifactId>                        <version>5.1.7.RELEASE</version>        </dependency><dependency>            <groupId>org.springframework.amqp</groupId>            <artifactId>spring-rabbit</artifactId>                    <version>2.1.8.RELEASE</version>        </dependency><dependency>            <groupId>junit</groupId>            <artifactId>junit</artifactId>            <version>4.12</version>        </dependency><dependency>            <groupId>org.springframework</groupId>            <artifactId>spring-test</artifactId>                        <version>5.1.7.RELEASE</version>        </dependency>    </dependencies>
</project>

配置整合

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans"       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"       xmlns:context="http://www.springframework.org/schema/context"       xmlns:rabbit="http://www.springframework.org/schema/rabbit"       xsi:schemaLocation="http://www.springframework.org/schema/beans       http://www.springframework.org/schema/beans/spring-beans.xsd       http://www.springframework.org/schema/context       https://www.springframework.org/schema/context/spring-context.xsd       http://www.springframework.org/schema/rabbit       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">    <!--加载配置文件-->    <context:property-placeholder location="classpath:properties/rabbitmq.properties"/><!-- 定义rabbitmq connectionFactory -->    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"                               port="${rabbitmq.port}"                               username="${rabbitmq.username}"                               password="${rabbitmq.password}"                               virtual-host="${rabbitmq.virtual-host}"/>    <!--定义管理交换机、队列-->    <rabbit:admin connection-factory="connectionFactory"/><!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机    默认交换机类型为direct，名字为：""，路由键为队列的名称    -->    <rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" autodeclare="true"/><!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" autodeclare="true"/><!--定义广播类型交换机；并绑定上述两个队列-->    <rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true"><rabbit:bindings>            <rabbit:binding queue="spring_fanout_queue_1"/>            <rabbit:binding queue="spring_fanout_queue_2"/>        </rabbit:bindings>    </rabbit:fanout-exchange><!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>    <!--定义广播交换机中的持久化队列，不存在则自动创建-->    <rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/><rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true">        <rabbit:bindings>            <rabbit:binding pattern="lxs.*" queue="spring_topic_queue_star"/>            <rabbit:binding pattern="lxs.#" queue="spring_topic_queue_well"/>            <rabbit:binding pattern="xzk.#" queue="spring_topic_queue_well2"/>        </rabbit:bindings>    </rabbit:topic-exchange><!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->    <rabbit:template id="rabbitTemplate" connectionfactory="connectionFactory"/>
</beans>

发送消息

创建测试文件 spring-rabbitmqproducer\src\test\java\com\lxs\rabbitmq\ProducerTest.java

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring/spring-rabbitmq.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 只发队列消息
* 默认交换机类型为 direct
* 交换机的名称为空，路由键为队列的名称
*/
@Test
public void queueTest(){
//路由键与队列同名
rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消
息。");
}/**
* 发送广播
* 交换机类型为 fanout
* 绑定到该交换机的所有队列都能够收到消息
*/
@Test
public void fanoutTest(){
/**
* 参数1：交换机名称
* 参数2：路由键名（广播设置为空）
* 参数3：发送的消息内容
*/
rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到
spring_fanout_exchange交换机的广播消息");
}
/**
* 通配符
* 交换机类型为 topic
* 匹配路由键的通配符，*表示一个单词，#表示多个单词
* 绑定到该交换机的匹配队列能够收到对应消息
*/
@Test
public void topicTest(){
/**
* 参数1：交换机名称
* 参数2：路由键名
* 参数3：发送的消息内容
*/
rabbitTemplate.convertAndSend("spring_topic_exchange", "lxs.bj", "发送到
spring_topic_exchange交换机lxs.bj的消息");
rabbitTemplate.convertAndSend("spring_topic_exchange", "lxs.bj.1", "发送
到spring_topic_exchange交换机lxs.bj.1的消息");
rabbitTemplate.convertAndSend("spring_topic_exchange", "lxs.bj.2", "发送
到spring_topic_exchange交换机lxs.bj.2的消息");
rabbitTemplate.convertAndSend("spring_topic_exchange", "xzk.cn", "发送到
spring_topic_exchange交换机xzk.cn的消息");
}
}

消费者依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.lxs</groupId>
<artifactId>spring-rabbitmq-consumer</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>5.1.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.1.8.RELEASE</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>5.1.7.RELEASE</version>
</dependency>
</dependencies>
</project>

配置整合

\2. 创建 spring-rabbitmq-consumer\src\main\resources\spring\spring-rabbitmq.xml 整合配置文件；

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!--加载配置文件-->
<context:property-placeholder
location="classpath:properties/rabbitmq.properties"/>
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}"
port="${rabbitmq.port}"
username="${rabbitmq.username}"
password="${rabbitmq.password}"
virtual-host="${rabbitmq.virtual-host}"/>
<bean id="springQueueListener"
class="com.lxs.rabbitmq.listener.SpringQueueListener"/>
<bean id="fanoutListener1"
class="com.lxs.rabbitmq.listener.FanoutListener1"/>
<bean id="fanoutListener2"
class="com.lxs.rabbitmq.listener.FanoutListener2"/>
<bean id="topicListenerStar"
class="com.lxs.rabbitmq.listener.TopicListenerStar"/>
<bean id="topicListenerWell"
class="com.lxs.rabbitmq.listener.TopicListenerWell"/>
<bean id="topicListenerWell2"
class="com.lxs.rabbitmq.listener.TopicListenerWell2"/>
<rabbit:listener-container connection-factory="connectionFactory" autodeclare="true">
<rabbit:listener ref="springQueueListener" queue-names="spring_queue"/>
<rabbit:listener ref="fanoutListener1" queuenames="spring_fanout_queue_1"/>
<rabbit:listener ref="fanoutListener2" queuenames="spring_fanout_queue_2"/>
<rabbit:listener ref="topicListenerStar" queuenames="spring_topic_queue_star"/>
<rabbit:listener ref="topicListenerWell" queuenames="spring_topic_queue_well"/>
<rabbit:listener ref="topicListenerWell2" queuenames="spring_topic_queue_well2"/>
</rabbit:listener-container>
<beans>

队列监听器

创建 spring-rabbitmqconsumer\src\main\java\com\lxs\rabbitmq\listener\SpringQueueListener.java

public class SpringQueueListener implements MessageListener{
public void onMessage(Message message) {
try {
String msg = new String(message.getBody(), "utf-8");
System.out.printf("接收路由名称为：%s，路由键为：%s，队列名为：%s的消息：%s \n",
message.getMessageProperties().getReceivedExchange(),
message.getMessageProperties().getReceivedRoutingKey(),
message.getMessageProperties().getConsumerQueue(),
msg);} catch (Exception e) {e.printStackTrace();}}
}

广播监听器

创建spring-rabbitmqconsumer\src\main\java\com\lxs\rabbitmq\listener\FanoutListener1.java

public class FanoutListener1 implements MessageListener {
public void onMessage(Message message) {
try {
String msg = new String(message.getBody(), "utf-8");
System.out.printf("广播监听器1：接收路由名称为：%s，路由键为：%s，队列名为：
%s的消息：%s \n",
message.getMessageProperties().getReceivedExchange(),
message.getMessageProperties().getReceivedRoutingKey(),
message.getMessageProperties().getConsumerQueue(),
msg);} catch (Exception e) {e.printStackTrace();}}
}

Spring Boot整合RabbitMQ

在Spring项目中，可以使用Spring-Rabbit去操作RabbitMQ https://github.com/spring-projects/spring-amqp

尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可，方便的使用RabbitTemplate发送消息，使用注解接收消息。

一般在开发过程中：

生产者工程：

application.yml文件配置RabbitMQ相关信息
在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定
注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机

消费者工程：

application.yml文件配置RabbitMQ相关信息
创建消息处理类，用于接收队列中的消息并进行处理

创建工程，添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
</parent>
<groupId>com.lxs</groupId>
<artifactId>springboot-rabbitmq-producer</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
</project>

启动类

package com.lxs.rabbitmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class);}
}

配置

package com.lxs.rabbitmq.config;
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
//交换机名称
public static final String ITEM_TOPIC_EXCHANGE =
"springboot_item_topic_exchange";
//队列名称
public static final String ITEM_QUEUE = "springboot_item_queue";
//声明交换机
@Bean("itemTopicExchange")
public Exchange topicExchange(){
return
ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();
}//声明队列
@Bean("itemQueue")
public Queue itemQueue(){
return QueueBuilder.durable(ITEM_QUEUE).build();
}
//绑定队列和交换机
@Bean
public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue,
@Qualifier("itemTopicExchange") Exchange
exchange){return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();}
}

测试

消费者端

创建模块，添加依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
</parent>
<groupId>com.lxs</groupId>
<artifactId>springboot-rabbitmq-consumer</artifactId>
<version>1.0-SNAPSHOT</version>
<dependencies><dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
</dependency>
</dependencies>
</project>

消息监听处理

RabbitMQ从入门到实战（图文并茂）相关推荐

慕课网_《RabbitMQ消息中间件极速入门与实战》学习总结
慕课网<RabbitMQ消息中间件极速入门与实战>学习总结时间:2018年09月05日星期三说明:本文部分内容均来自慕课网.@慕课网:https://www.imooc.com 教学源 ...
《 Docker 技术入门与实战》读书笔记 ( CentOS 安装 Docker )
前些天发现了一个巨牛的人工智能学习网站,通俗易懂,风趣幽默,忍不住分享一下给大家.点击跳转到教程. PS :个人所有读书笔记只记录个人想要的内容,很可能原书大量内容没有纳入笔记中... ... 以下全 ...
Google Android开发入门与实战
Google Android开发入门与实战 [作者]靳岩;姚尚朗 [同作者作品] [作译者介绍] [出版社] 人民邮电出版社 [书号] 9787115209306 [上架时间] 2 ...
Google Android开发入门与实战(china-pub首发免运费)
Google Android开发入门与实战(china-pub到货首发免运费) [作者]靳岩;姚尚朗 [同作者作品] [作译者介绍] [出版社] 人民邮电出版社 [书号] 97871 ...
Docker技术入门与实战（第2版）.
容器技术系列 Docker技术入门与实战第2版杨保华戴王剑曹亚仑编著图书在版编目(CIP)数据 Docker技术入门与实战 / 杨保华,戴王剑,曹亚仑编著. -2版. -北京:机械工业出版 ...
python3破冰人工智能pdf 微盘_Python 3破冰人工智能从入门到实战
<Python 3破冰人工智能从入门到实战>简介: 内容简介: 本书创新性地从数学建模竞赛入手,深入浅出地讲解了人工智能领域的相关知识.本书内容基于Python 3.6,从人工智能领域的 ...
Redis入门到实战
redis入门与实战一.Nosql概述 1.为什么要用Nosql 1.1 单机 MySQL 的美好时代来源博客(https://www.cnblogs.com/lukelook/p/1113520 ...
《ClickHouse入门、实战与进阶》的创作之路
写作不是思考的记录,写作就是思考本身. --理查德·费曼目录写作的重要性写作之路 OLAP技术对于企业决策者.数据分析师等至关重要分享一些经验最后写作的重要性本文开头借用了费曼的名言来表 ...
Kubernetes权威指南第2版和 Docker技术入门与实战第2版两本容器的书下载地址
两本书的下载链接下载链接链接:https://pan.baidu.com/s/13gv0ZQRiHfvLwgwjsvUiEA 密码:hi8o Kubernetes权威指南第2版目录: 第1章 K ...

RabbitMQ从入门到实战（图文并茂）

RabbitMQ从入门到实战（图文并茂）相关推荐

最新文章

热门文章