RabbitMq使用与整合

MQ基本概念

MQ概述

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

（队列是一种容器，用于存放数据的都是容器，存放消息的就是消息队列）

分布式系统的调用：

方式一：直接调用

方式二：间接调用

A将数据存放到中间一个系统，通过中间的系统发送到B

中间系统可以成为中间件MQ

MQ是用于存放消息的中间件

被调用者叫生产者调用者是消费者

MQ的优势和劣势

优势

应用解耦：提高系统容错性和可维护性。

异步提速：提升用户体验和系统吞吐量。

削峰填谷：提高系统稳定性。

应用解耦

系统的耦合性越高，容错性就越低，可维护性就越低。

使用 MQ 使得应用间解耦，提升容错性和可维护性

异步提速

提升用户体验和系统吞吐量（单位时间内处理请求的数目）。

削峰填谷（削峰）

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

使用MQ后，可以提高系统稳定性。

劣势

系统可用性降低

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

系统复杂度提高

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

一致性问题

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

既然 MQ 有优势也有劣势，那么使用 MQ 需要满足什么条件呢？

消费者-->生产者

1.生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

2.容许短暂的不一致性。

3.确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

RabbitMQ基本介绍

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

RabbitMQ 基础架构

Broker 中间者服务

procedure 和consumer都是客户端

客户端通过链接和服务端进行通信所以需要建立起来连接然后进行通信a

使用channel（管道）节省资源

一个rabbitmq里面有很多的虚拟机相当于mysql里面有很多数据库，数据库里面有很多表，都是独立的。

每个虚拟机里面有很多的exchange和queue 独立分区的作用

RabbitMQ 中的相关概念

Broker：接收和分发消息的应用，RabbitMQ Server就是 Message Broker。

Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等。

Connection：publisher／consumer 和 broker 之间的 TCP 连接。

Channel：如果每一次访问 RabbitMQ 都建立一个 Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel 是在 connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的 channel 进行通讯，AMQP method 包含了channel id 帮助客户端和message broker 识别 channel，所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销。

Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：

        direct (point-to-point)

        topic (publish-subscribe)

        fanout (multicast)

Queue：消息最终被送到这里等待 consumer 取走

Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据

RabbitMQ的6 种工作模式

RabbitMQ 提供了 6 种工作模式：

简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式（远程调用，不太算 MQ；暂不作介绍）。

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

RabbitMQ的安装和配置

安装依赖环境

在线安装依赖环境：

yum -y install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

安装Erlang

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm

安装RabbitMQ

#安装依赖的包

rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

#安装rabbitmq

rpm -ivh rabbitmq-server-3.7.18-1.el7.noarch.rpm

rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm socat-1.7.3.2-2.el7.x86_64.rpm rabbitmq-server-3.7.18-1.el7.noarch.rpm

启动RabbitMQ

systemctl start rabbitmq-server # 启动服务

systemctl stop rabbitmq-server # 停止服务

systemctl restart rabbitmq-server # 重启服务

systemctl status rabbitmq-server # 查看状态

开启管理界面及配置

# 开启管理界面

rabbitmq-plugins enable rabbitmq_management

# 修改默认配置信息

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.7.18/ebin/rabbit.app

# 比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest

修改之后重启一下rabbitmq

入门实例

1.添加虚拟主机

2.添加用户

3.重新设置权限

点击虚拟主机设置权限

4.idea连接

项目结构搭建

mq 导入依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version>
</dependency>

生产者

public static void main( String[] args ) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setUsername("root");connectionFactory.setPassword("root");connectionFactory.setHost("192.168.229.16");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/test");
//        建立连接Connection connection = connectionFactory.newConnection();
//        管道Channel channel = connection.createChannel();
//        创建队列/** String queue 队列的名称* boolean durable 持久化* boolean exclusive 是否独占* boolean autoDelete 是否自动删除* Map<String,Object> arguments 参数* */channel.queueDeclare("test01",false,false,false,null);
//        发布消息channel.basicPublish("","test01",null,"第一次发送".getBytes());}

消费者

{public static void main( String[] args ) throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setUsername("root");connectionFactory.setPassword("root");connectionFactory.setHost("192.168.229.16");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/test");
//        建立连接Connection connection = connectionFactory.newConnection();
//        管道Channel channel = connection.createChannel();
//        消费信息Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String s = new String(body);System.out.println(s);}};channel.basicConsume("test01",true,consumer);}

同时消费后消息组内为 0

RabbitMQ工作模式

Work queues工作队列模式

Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

代码

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

1.复制一个消费者

2.先运行起两个消费者

3.在生产者中多发布几条消息

for (int i = 0; i < 10; i++) {channel.basicPublish("","test01",null,("第"+i+"次发送").getBytes());
}

4.两个消费者会采用轮询的方式拿到消息

在一个队列中如果有多个消费者，那么消费者之间对于同一个消息的关系是竞争的关系

订阅模式类型

而在订阅模型中，多了一个exchange角色，而且过程略有变化：

生产者发消息给交换机，交换机将消息路由分发给队列，消费者监听队列接收信息

Exchange：交换机。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有常见以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

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

代码实现

生产者更改代码逻辑为:

//        创建交换机channel.exchangeDeclare("Exchange", BuiltinExchangeType.FANOUT,false);
//        创建队列channel.queueDeclare("test01Ex",false,false,false,null);channel.queueDeclare("test02Ex",false,false,false,null);
//        队列绑定交换机channel.queueBind("test01Ex","Exchange","");channel.queueBind("test02Ex","Exchange","");
//        发布消息for (int i = 0; i < 10; i++) {channel.basicPublish("Exchange","",null,("第"+i+"次发送").getBytes());}

消费者分别从两个队列获取消息

Routing路由模式(Direct：定向)

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey(路由key）

消息的发送方在向Exchange发送消息时，也必须指定消息的RoutingKey

Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey进行判断，只有队列的RoutingKey与消息的 RoutingKey完全一致，才会接收到消息

 channel.queueBind(queuename,exchangename,"error"); // errorchannel.queueBind(queuename2,exchangename,"error");// error info channel.queueBind(queuename2,exchangename,"info");// 发送消息//String exchange, String routingKey, BasicProperties props, byte[] bodychannel.basicPublish(exchangename,"info",null,"hello".getBytes());

Topics通配符模式

Topic与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定RoutingKey的时候使用通配符！

RoutingKey一般都是有一个或多个单词组成，多个单词之间以”.”分割

#：匹配一个或多个词

*：匹配不多不少恰好1个词 test.* test.insert

channel.queueBind(queuename,exchangename,"order.*");channel.queueBind(queuename,exchangename,"*.error");channel.queueBind(queuename2,exchangename,"*.*");//发送消息//String exchange, String routingKey, BasicProperties props, byte[] bodychannel.basicPublish(exchangename,"goods.info",null,"hello".getBytes());

模式总结

RabbitMQ工作模式：

`简单模式 HelloWorld`

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

工作队列模式 Work Queue

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）

`发布订阅模式 Publish/subscribe`

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列

`路由模式 Routing`

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

`通配符模式 Topic`

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的routing key，当发送消息到交换机后，交换机会根据routing key将消息发送到对应的队列

SpringBoot整合Mq

在Spring项目中，可以使用Spring-Rabbit去操作RabbitMQ

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

创建工程结构

添加依赖(sys-mq）

<parent><groupId>com.example</groupId><artifactId>mqdemo02</artifactId><version>0.0.1-SNAPSHOT</version></parent><artifactId>sys-mq</artifactId><packaging>pom</packaging><name>sys-mq</name><url>http://maven.apache.org</url><modules><module>mq-product</module><module>mq-consumer</module></modules><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency>
<!--        rabbitmq--><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>

sys-mq里面两个子模块的 application.yml

server

port: 记得修改

spring:rabbitmq:username: rootpassword: roothost: 192.168.229.16virtual-host: /testport: 5672
server:port: 8086

生产者

@Configuration
public class RabbitMqConfig {
//    设置交换机的名称和队列的名字public static final String EXCHANGE_NAME="exchange_topic-test";public static final String QUEUE_NAME="queue_topic-test";public static final String QUEUE_NAME2="queue_topic-test2";
//    创建交换机 将交换机作为bean注入到spring中@Bean("topicExchange")public Exchange topicExchange(){return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(false).build();}
//    队列@Bean("topicQueue")public Queue topicQueue(){return QueueBuilder.durable(QUEUE_NAME).build();}@Bean("topicQueue2")public Queue topicQueue2(){return QueueBuilder.durable(QUEUE_NAME2).build();}
//    将队列与交换机进行绑定@Beanpublic Binding exchangeWithQueue(@Qualifier("topicExchange") Exchange exchange,@Qualifier("topicQueue") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("*.test").noargs();}@Beanpublic Binding exchangeWithQueue2(@Qualifier("topicExchange") Exchange exchange,@Qualifier("topicQueue2") Queue queue) {return BindingBuilder.bind(queue).to(exchange).with("test.#").noargs();}
}

@SpringBootTest
public class MyTest {public static final String EXCHANGE_NAME="exchange_topic-test";@Resourceprivate RabbitTemplate rabbitTemplate;@Testvoid sendMsg() {rabbitTemplate.convertAndSend(EXCHANGE_NAME,"success.test","测试整合");}
}

消费者

@Component
public class MyListener {
//    监听队列的消息@RabbitListener(queues = "queue_topic-test")public void listenQueue(Message message) {byte[] body = message.getBody();System.out.println(new String(body));}
}