RabbitMQ消息模型之Simple消息模型

simple消息模型

生产者

package com.example.demo02.mq.simple;

import com.example.demo02.mq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;

/**
 * @author Allen
 * 4/10/2024 8:07 PM
 * @version 1.0
 * @description: 简单模型-生产者
 */
public class SimpleSender {
    public static void main(String[] args) throws Exception {
        // 1、创建连接
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道
        Channel channel = connection.createChannel();
        // 3、声明队列：队列可以缓存数据、mq的交换机不能存储数据 simple模型使用的是默认交换机
        // 参数1：队列名称 参数2：是否持久化 参数3：是否排他性 参数4：是否自动删除 参数5：其他属性
        channel.queueDeclare("simple_queue", false, false, false, null);
        // 4、发送消息：发送到队列中
        // 在simple模型中   参数1：交换机 参数2：队列名称 参数3：传递消息额外设置 参数4：消息的具体内容(任意数据类型)
        String msg = "hello rabbitmq";
        channel.basicPublish("", "simple_queue", null, msg.getBytes());
        // 5、关闭通道
        channel.close();
        // 6、关闭连接
        connection.close();
    }
}
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

消费者

package com.example.demo02.mq.simple;

import com.example.demo02.mq.util.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * @author Allen
 * 4/10/2024 8:59 PM
 * @version 1.0
 * @description: 简单模型-消费者
 */
public class SimpleReciver {
    public static void main(String[] args) throws Exception {
        // 1、创建连接
        Connection connection = ConnectionUtils.getConnection();
        // 2、创建通道
        Channel channel = connection.createChannel();
        // 3、声明队列：如果已声明可以忽略
        // 4、监听队列：
        channel.queueDeclare("simple_queue", false, false, false, null);
        Consumer consumer = new DefaultConsumer(channel){
            // 消息的具体处理逻辑
            // 消费者接收到消息后调用此方法
            // 如果AutoAck为true，此方法一旦被调用，消息会被确认
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者接收到消息："+new String(body));
                /*
                channel.basicReject();    //拒绝消息，可以丢弃消息，也可以重新入队
                channel.basicNack();      //不确认消息，可以设置是否重新入队
                */
                channel.basicAck(envelope.getDeliveryTag(),false);         //手动确认消息，参数1：消息的标识，参数2：是否开启多个消息同时确认
            }
        };
/*        参数1：队列名称 参数2：是否自动确认 参数3：消费者
        队列名称：消费哪个队列的消息
        AutoAck：是否自动确认，如果为true，表示消息一旦被接收，自动向mq回复接收到，mq会删除消息，如果为false，需要手动确认，如果消费者挂掉，消息会被重回队列
        Consumer：消费者对象
        channel.basicConsume("simple_queue", true,consumer);
        避免消息丢失，手动确认
*/
        channel.basicConsume("simple_queue", false,consumer);

    }
}
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

Unacked ： 消费者用了、但是没有告诉交换机、就是未确认的状态

当消息被消费了，且告诉交换机了，交换机就会将这个消息删除