前言

🍊作者简介： 不肯过江东丶，一个来自二线城市的程序员，致力于用“猥琐”办法解决繁琐问题，让复杂的问题变得通俗易懂。
🍊支持作者： 点赞👍、关注💖、留言💌~

在上篇博客中我们讲解了如何在 Win10 环境下安装部署 RabbitMQ，相信各位小伙伴已经搭建出了一套自己的 RabbitMQ 环境，那么今天就接着往后说，讲讲 Java 代码对 RabbitMQ 的简单操作。

传送门：大聪明教你学Java | Win10 环境下安装部署 RabbitMQ

Java 代码对 RabbitMQ 的简单操作

RabbitMQ 的工作原理及基本概念

在讲代码之前，我们先简单了解一下 RabbitMQ 的工作原理（图片来自百度图片，侵删~）👇

我们谈到消息队列，通常都会想到这其中的三个角色：生产者、消费者
、消息队列。生产者将消息发送到消息队列，消费者从消息队列中获取消息进行处理。但是对于 RabbitMQ 而言，它在此基础上做了一层抽象，引入了交换器 Exchange 的概念，交换器是作用于生产者和消息队列之间的中间桥梁，它起了一种消息路由的作用，也就是说生产者并不和消息队列直接关联，而是先讲消息发送给交换器，再由交换器路由到对应的消息队列。通过上图我们可以看到，生产者并没有直接将消息发送给消息队列，而是通过建立与 Exchange（交换器）的 Channel（信道），将消息发送给 Exchange，Exchange 根据路由规则，将消息转发给指定的消息队列。消息队列储存消息，等待消费者取出消息，消费者通过建立与消息队列相连的 Channel，从消息队列中获取消息。

下面我们再简单的介绍一下 RabbitMQ 中的一些基本概念：

1. Channel（信道）：多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的 TCP 连接内的虚拟连接，复用 TCP 连接的通道。
2. Producer（生产者）：向消息队列发布消息的客户端应用程序。
3. Consumer（消费者）：从消息队列取得消息的客户端应用程序。
4. Queue（消息队列）：存储消息的一种数据结构，它可以用来保存消息，直到消息发送给消费者。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将消息取走。需要注意的是，当多个消费者订阅同一个 Queue，这时 Queue 中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，每一条消息只能被一个订阅者接收。
5. Exchange（交换器）：它用来接收生产者发送的消息并将这些消息按照路由规则转发到消息队列。交换器用于转发消息，并不会存储消息 ，如果没有消息队列（Queue）绑定到交换器（Exchange）的话，它会直接丢弃掉生产者发送过来的消息。

Java 代码对 RabbitMQ 的简单操作

通过上面的内容，让我们对 RabbitMQ 的工作原理及基本概念有了一个基本的了解，接下来我们就一起看看如何用 Java 代码操作 RabbitMQ。

首先我们还是先引入 Maven 依赖👇

<!-- rabbitmq 依赖 -->
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.8.0</version>
</dependency>

<!-- io依赖 -->
<dependency>
    <groupId>commons-io</groupId>
    <artifactId>commons-io</artifactId>
    <version>2.6</version>
</dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13

接下来创建一个生产者类，向 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;

/**
 * RabbitMQ 生产者
 * @description: Producer
 * @author: 庄霸.liziye
 * @create: 2022-06-22 10:36
 **/
public class Producer {

    public static final String QUEUE_NAME = "Hello";

    public static void main(String[] args) throws IOException, TimeoutException {
        //创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //RabbitMQ 安装的服务器ip地址
        factory.setHost("127.0.0.1");
        //用户名
        factory.setUsername("guest");
        //密码
        factory.setPassword("guest");
        //创建连接
        Connection connection = factory.newConnection();
        //获取信道
        Channel channel = connection.createChannel();
        /**
         * 生成一个队列
         * 参数：
         * 1 队列名称
         * 2 队列里的消息是否持久化，默认情况消息存储在内存中
         * 3 该队列是否供多个消费者进行消费，是否进行消息共享
         * 4 是否自动删除
         * 5 其他参数
         */
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //发消息
        String message = "hello, I am the Producer!";
        /**
         * 发送一个消费
         * 参数：
         * 1 发送到哪个交换机
         * 2 路由的key值 本次是队列名称
         * 3 其他参数信息
         * 4 发送消息的消息体
         */
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("消息发送完毕");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

接下来我们运行代码，会看到控制台输出了“消息发送完毕”的提示文字👇

下面我们再登录一下 RabbitMQ 管理页面，就会在 Queue 选项卡下看到我们刚刚建立的消息队列👇

接下来我们再创建一个消费者，从消息队列中把消息取出来👇

import com.rabbitmq.client.*;

/**
 * RabbitMQ 消费者
 * @description: Consumer
 * @author: 庄霸.liziye
 * @create: 2022-06-22 10:58
 **/
public class Consumer {

    public static final String QUEUE_NAME = "Hello";

    public static void main(String[] args) throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        //工厂IP，连接rabbitmq的队列
        factory.setHost("127.0.0.1");
        //用户名
        factory.setUsername("guest");
        //密码
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        System.out.println("等待接收消息....");

        /**
         * 消费者接受消息 channel.basicConsume()
         * 参数：
         * 1 消费哪个队列
         * 2 消费成功后是否要自动应答
         * 3 消费者未成功消费的回调
         * 4 消费者取录消费的回调
         */
        //推送的消息如何进行消费的接口回调
        DeliverCallback deliverCallback=(consumerTag, delivery)->{
            String message= new String(delivery.getBody());
            System.out.println(message);
        };
        //取消消费的一个回调接口 如在消费的时候队列被删除掉了
        CancelCallback cancelCallback=(consumerTag)->{
            System.out.println("消息消费被中断");
        };

        channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

我们运行一下消费者的代码，就可以看到刚刚发送到消息队列中的消息被成功取出来了✌

小结

本人经验有限，有些地方可能讲的没有特别到位，如果您在阅读的时候想到了什么问题，欢迎在评论区留言，我们后续再一一探讨🙇‍

希望各位小伙伴动动自己可爱的小手，来一波点赞+关注 (✿◡‿◡) 让更多小伙伴看到这篇文章~ 蟹蟹呦(●’◡’●)

如果文章中有错误，欢迎大家留言指正；若您有更好、更独到的理解，欢迎您在留言区留下您的宝贵想法。

你在被打击时，记起你的珍贵，抵抗恶意；
你在迷茫时，坚信你的珍贵，抛开蜚语；
爱你所爱 行你所行 听从你心 无问东西