RabbitMQ快速入门

MQ简介

1、同步通信

图片

2、异步通信

图片

RabbitMQ快速上手

基本介绍：

Producer和Consumer

Connection和Channel

Virtual host

Queue

Exchange

工作流程

AMQP

Java编写RabbitMQ生产者消费者

生产者

1.建立连接

2.开启信道

3.声明交换机

4.声明队列

5.发送消息

6.资源释放

生产者全部代码：

消费者

执行结果

MQ简介

MQ(Message Queue)消息队列，本质上是队列，满足队列FIFO(先入先出)的性质,队列中存放的内容是消息，消息可以是只包含文本字符串，JSON，也可以是对象等。

MQ经常用于分布式系统之间的通信，系统之间的通信通常有两种方式：

1、同步通信

直接调用对方的服务，数据从一端出发到达另一端。

图片

2、异步通信

数据从⼀端发出后，先进⼊⼀个容器进⾏临时存储，当达到某种条件后，再由这个容器发送给另⼀端，这个容器通常就是MQ(message queue)。

图片

RabbitMQ快速上手

基本介绍：

RabbitMQ是MQ的一种实现，工作流程如下：

RabbitMQ是一个消息中间件，也是一个生产者消费者模型，负责接收，存储并转发消息，消息传递的过程类似于发快递，你把快递放到驿站，快递小哥帮你把快递送到接收人的手上。这个过程中，你就相当于Producer，接收人相当于Consumer，快递站就是RabbitMQ。

Producer和Consumer

Producer(生产者)，是RabbitMQ的客户端，用于发送消息到RabbitMQ。

Consumer(消费者)，也是RabbitMQ的客户端，用于向RabbitMQ接收消息。

Broker(代理)，RabbitMQ服务器节点，用于接收和发送消息。

生产者发送消息到RabbitMQ服务器，让RabbitMQ进行路由转发到对应的消费者。

生产者发送的消息带有标签，代理（RabbitMQ服务器）会根据标签路由，找到需要的消费者，进行消息转发。消费者接收消息进行消费过程中标签就会被丢掉，也就是说消费者并不会知道消息的发送者是谁。

Connection和Channel

Connection(连接)：客户端和RabbitMQ服务器之间的⼀个TCP连接，建立连接后才可以进行消息的发送和接收。

Channel(信道)：类似于发送消息的一个通道，每个TCP连接可以有多个信道，每个信道都是独立的虚拟连接，消息的发送和接收都是基于Channel(信道)的。

Virtual host

Virtual host：虚拟主机，给消息队列提供逻辑上的隔离，一个BrokerServer可以有多个虚拟主机，当不同用户使用RabbitMQ Server提供的服务时，可以使用划分多个虚拟主机的方式将一系列业务隔离开来，类似于MySQL中不同的数据库。

Queue

Queue：队列，用于存放消息，一个队列可以被多个消费者订阅。

Exchange

Exchenge：交换机，交换机负责接收生产者发送的消息，并按照规则将消息路由到队列上。类似于快递小哥将快递送给

工作流程

Proucer生产消息 -> Producer连接到RabbitMQBroker，建立连接(Connection)，开启信道(Channel) -> Producer声明交换机(Exchange),路由消息 ->Producer声明队列(Queue) -> Produce发送消息到RabbitMQBroker-> RabbitMQBroker接收消息，并放入相应的队列(Queue)中，未找到相应队列根据生产者的配置，选择丢弃或者退回给生产者。

AMQP

AMQP是一种高级消息队列协议，定义了一套确定的消息交换功能，包括交换机，队列等，这些组件共同工作，使生产者能够将消息发送到交换机，由队列接收并等待消费者接收。AMQP还定义了一个网络协议，允许客户端应用通过这个协议与消息代理和AMQP模型进行交互通信。RabbitMQ使AMQP协议的Erlang实现。

Java编写RabbitMQ生产者消费者

打开管理页面，添加用户：

这里添加的测试用户名称为：test 密码 123456。

创建虚拟机：

这里添加的虚拟机名称为：test

返回admin页面，点击刚刚创建的test用户，添加虚拟机操作权限。

进入IDEA创建一个Maven项目，导入依赖：

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

生产者

1.建立连接

使用依赖中的ConnectionFactory类创建连接，设置关键信息：

ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("139.9.116.124");
connectionFactory.setPort(5672);
connectionFactory.setUsername("fwx");
connectionFactory.setPassword("123456");
connectionFactory.setVirtualHost("测试");
Connection connection = connectionFactory.newConnection();

2.开启信道

Channel channel = connection.createChannel();

3.声明交换机

这里使用的是RabbitMQ默认提供的交换机，不需要代码进行声明

4.声明队列

使用channel的queueDeclare方法，依此填入相关参数：

channel.queueDeclare("hello",true,false,false,null);

打开源码，查看参数的含义：

第一个参数queue代表队列的名称，第二个参数durable代表是否持久化（true-设置队列为持久化, 待久化的队列会存盘,服务器重启之后, 消息不丢失），第三个参数exclusive代表是否独占（是否独占, 只能有⼀个消费者监听队列），第四个参数autoDelete代表是否自动删除（没有消费者自动删除队列），第五个参数arguments代表一些配置的参数。

5.发送消息

使用Channel的basicPublish()方法进行消息的发送，这里发送的消息为"hello RabbitMQ~"，消息发送10次：

for (int i = 0; i < 10; i++) {String msg = "hello RabbitMQ~";channel.basicPublish("","test1",null,msg.getBytes());
}

打开方法，分析方法参数：

第一个参数exchange代表交换机的名称(这里使用默认交换机"")，第二个参数routingKey代表路由的名称（使用默认交换机，路由名称要和队列名称相同才可以找到对应队列），第三个参数body代表发送的消息体（将String类型转换为字节数组）。

6.资源释放

channel.close();
connection.close();

最后将资源关闭即可。

生产者全部代码：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立链接ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址connectionFactory.setHost("139.9.116.124");//端口号connectionFactory.setPort(5672);//用户名connectionFactory.setUsername("test");//密码connectionFactory.setPassword("123456");//虚拟机connectionFactory.setVirtualHost("test");//连接建立Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明交换机,使用内置交换机//4.声明队列channel.queueDeclare("test1",true,false,false,null);//5.发送消息for (int i = 0; i < 10; i++) {String msg = "hello RabbitMQ~";channel.basicPublish("","test1",null,msg.getBytes());}//6.资源释放channel.close();connection.close();}
}

消费者

消费者与生产者逻辑相似，这里不再讲解，全部代码：

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.建立链接ConnectionFactory connectionFactory = new ConnectionFactory();//主机地址connectionFactory.setHost("139.9.116.124");//端口号connectionFactory.setPort(5672);//用户名connectionFactory.setUsername("test");//密码connectionFactory.setPassword("123456");//虚拟机connectionFactory.setVirtualHost("test");//连接建立Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明队列(生产者已经声明)//4.消费消息/**参数声明* basicConsume(String queue, boolean autoAck, Consumer callback)* queue:队列名* autoAck:自动确认* callback：接收到消息执行逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息，就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//TODOSystem.out.println("接收的消息："+new String(body));}};channel.basicConsume("test1",true,consumer);//等待程序执行完成Thread.sleep(2000);//5.释放资源channel.close();connection.close();}
}

生产者已经声明过队列，消费者无需声明，消费者通过重写handleDelivery()方法（这个方法在接收消息时自动执行）去消费消息。