RabbitMQ工作流程及使用方法

news2025/7/19 18:30:33

一、什么是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> 运行生产者代码：