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

RabbitMQ系列学习笔记（二）--简单模式

article 2024/10/21 10:55:57

文章目录

一、简单模式原理
二、简单模式实战
- 1、添加依赖
- 2、生产者代码
- 3、消费者代码
- 4、简单模式流程总结

本文参考：
尚硅谷RabbitMQ教程丨快速掌握MQ消息中间件rabbitmq
RabbitMQ 详解
Centos7环境安装Erlang、RabbitMQ详细过程(配图)

一、简单模式原理

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序。
C：消费者：消息的接受者，会一直等待消息到来。
queue：消息队列，图中红色部分。类似一个箱子，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

简单模式的特点：

一个生产者对应一个消费者，通过队列进行消息传递。
该模式使用 direct 交换机，direct 交换机是 RabbitMQ 默认的交换机。

二、简单模式实战

代码部分使用java进行实验，版本jdk1.8，基于maven环境运行。

1、添加依赖

首先创建maven工程并添加依赖：

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

2、生产者代码

在简单模式下，涉及到的部分只有生产者、消费者和队列，生产者负责向队列中发送消息，这个过程首先需要创建连接工厂（factory），设置IP地址、用户名密码等必要参数，其次需要建立连接（connection）当连接建立完成后，在连接中开通信道（channel），通过信道将消息发送（publish）到队列，最后发送完毕以后要关闭连接。基本过程明确以后，以下是代码部分。

/**
 * Description: 简单模式生产者
 */
public class Producer {
    
    //队列名称
    private static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {
        // 1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.222.139");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("scr1pt_yang.");
        factory.setVirtualHost("/");
        // 2.创建连接
        Connection connection = factory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();
        
        // 4.创建队列，如果已经存在，则直接使用
        /**
         * 方法参数:
         * 参数1: 队列名称
         * 参数2: 是否持久化。true表示队列会保存磁盘，MQ重启后该队列还在
         * 参数3: 是否私有化。true表示只有声明它的连接才能访问，false表示所有消费者都能访问
         * 参数4: 是否自动删除。true表示不再使用队列时自动删除队列
         * 参数5: 队列其他参数。如：x-message-ttl,x-expires等，null表示无其他参数提供
         */
        channel.queueDeclare(QUEUE_NAME, false, false ,false, null);
        // 5.发送消息
        String message = "hello world!";
        /**
         * 方法参数:
         * 参数1: 交换机名，""表示使用默认的交换机
         * 参数2: 路由key，简单模式为队列名
         * 参数3: 消息的其他参数，如路由头等，null表示没有其他参数信息
         * 参数4: 消息体（字节数组格式），需要转二进制发送
         */
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        // 6.关闭信道和连接
        channel.close();
        connection.close();
        System.out.println("消息发送成功！！！");
    }
}

3、消费者代码

消费者创建工厂建立连接最终拿到channel的过程与生产者一致。生产者发送使用的是channel.basicPublish()，消费者接收消息或者说是消费消息使用的是channel.basicConsume()，针对获取到消息以后的下一步执行操作，可以采用回调函数deliverCallback的方式，也可以采用重写DefaultConsumer的handleDelivery方法，从body中接收message，并对消息进行下一步操作。

/**
 * Description: 简单模式消费者
 */
public class Consumer {

    //队列名称
    private static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {
        // 1.创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.222.139");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("scr1pt_yang.");
        factory.setVirtualHost("/");
        // 2.创建连接
        Connection connection = factory.newConnection();
        // 3.建立信道
        Channel channel = connection.createChannel();
        System.out.println("等待接收消息....");
        // 4.监听队列
        /**
         * 方法参数:
         * 参数1: 监听的队列名
         * 参数2: 是否自动应答。如果为false，则需要手动确认消息已收到，否则MQ会一直发送消息
         * 参数3: Consumer的实现类，重写该类方法表示接收到消息后如何消费
         */
        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body);
                System.out.println("接收到的消息为: " + message);
            }
        });
    }
}