RabbitMQ工作流程及使用方法

一、什么是RabbitMQ

RabbitMQ 是一款基于 ‌AMQP（高级，消息队列协议）‌ 的开源消息中间件，专为分布式系统设计，用于实现应用程序间的异步通信，其核心功能是通过 ‌消息代理（Message Broker）‌ 机制，实现可靠的消息传递、存储和路由，支持跨语言和跨平台交互。

 二、RabbitMQ的工作流程

1. Producer 和 Broker 建立一个连接（Connection）并声明一个信道（channel）；

2. Producer 声明 一个交换机（Exchange）；

3. Producer 声明一个队列（Queue）；

4. Producer 与 Broker 连接建立完毕，开始向 Broker 传输信息。

三、RabbitMQ使用方法（入门）

3.1 引入依赖

创建一个maven项目并添加RabbitMQ依赖：

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

3.2 创建实体类ProducerDemo，ConsumerDemo

package rabbitmq;

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 ProducerDemo {
public static void main(String[] args) throws IOException, TimeoutException {
/*
* 根据rabbitmq的工作流程图，主要工作分为4步：
* 1.服务器ip
* 2.服务器端口号
* 3.账号、密码
* 4.虚拟主机*/

//1.建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("110.41.17.130");
connectionFactory.setPort(5672);
connectionFactory.setPassword("study");
connectionFactory.setUsername("study");
connectionFactory.setVirtualHost("java113");
Connection connection = connectionFactory.newConnection();

//2.开启信道
Channel channel = connection.createChannel();
//3.声明交换机(使用内置交换机）
//4.声明队列
/*AMQP.Queue.DeclareOk queueDeclare(String var1, boolean var2, boolean var3, boolean var4, Map<String, Object> var5) throws IOException;
*
*参数说明：
* s: 队列名
* b: 可持久化
* b1: 是否独占（ 是否只能有一个消费者读取队列中的信息）
* b2: 没有消费者时是否自动删除
* map: 参数*/
channel.queueDeclare("hello",true,false,false,null);
//5.发送消息
/*
* void basicPublish(String var1, String var2, AMQP.BasicProperties var3, byte[] var4) throws IOException;
* 参数说明：
* var1 交换机名称
* var2 内置交换机，和队列名称一致（路由规则）
* var3 属性配置
* var4 消息*/
for (int i = 0; i < 10; i++) {
String msg = "hello rabbitmq " + i;
channel.basicPublish("","hello",null,msg.getBytes());
}
System.out.println("消息发送成功");
//6.进行资源释放
channel.close();
connection.close();
}
}

package rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerDemo {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

//1.创建连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("110.41.17.130");
connectionFactory.setPort(5672);
connectionFactory.setUsername("study");
connectionFactory.setPassword("study");
connectionFactory.setVirtualHost("java113");
Connection connection = connectionFactory.newConnection();
//2.创建channel
Channel channel = connection.createChannel();
//3.声明一个队列（如果生产者声明了可以省略）
channel.queueDeclare("hello",true,false,false,null);
//4.消费消息
/*
* 参数说明：
* 1. 队列名称
* 2. 是否自动确认(即确认consumer是否收到消息）
* 3. 接收到消息后，执行的逻辑*/
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//TODO
System.out.println("接收消息" + new String(body));
}
};
channel.basicConsume("hello",true,consumer);

Thread.sleep(1000);
//5.释放资源
channel.close();
connection.close();
}
}

1> 运行生产者代码：

查看RabbitMQ管理界面的队列信息：

可以看到，队列中新增了10条信息

2>运行消费者代码

查看队列信息：

可以看到，队列中的信息已经被消费完毕

四、总结

在RabbitMQ的使用中，主要有以下几步：

1> 建立连接：无论是生产者还是消费者，要与服务器建立连接，需要知道：

       （1）服务器的 IP 地址和 端口号；

       （2）用户名 和 密码；

       （3）要连接到服务器的哪台虚拟机。

2>创建信道（channel）

3>声明交换机（使用内置交换机可不用声明）

4>声明队列（如果生产者已经声明队列，消费者可以不声明（不建议））

5>生产者使用 channel.basicPublish 方法发送信息，消费者使用chaanel.basicConsumer 方法消费信息