当前位置：首页 > ds >正文

RabbitMQ工作流程及使用方法

ds 2025/7/5 16:54:05

一、什么是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.创建channelChannel channel = connection.createChannel();//3.声明一个队列（如果生产者声明了可以省略）channel.queueDeclare("hello",true,false,false,null);//4.消费消息/** 参数说明：* 1. 队列名称* 2. 是否自动确认(即确认consumer是否收到消息）* 3. 接收到消息后，执行的逻辑*/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("hello",true,consumer);Thread.sleep(1000);//5.释放资源channel.close();connection.close();}
}

1> 运行生产者代码：