RabbitMQ-消息模型

准备条件

Maven.xml

1.  
2.<project xmlns="http://maven.apache.org/POM/4.0.0"  
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
4.         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">  
5.    <modelVersion>4.0.0modelVersion>  
6.  
7.    <groupId>com.lhq.rabbitmqgroupId>  
8.    <artifactId>rabbitmq-helloartifactId>  
9.    <version>1.0-SNAPSHOTversion>  
10.  
11.    <properties>  
12.        <maven.compiler.source>8maven.compiler.source>  
13.        <maven.compiler.target>8maven.compiler.target>  
14.    properties>  
15.  
16.    <dependencies>  
17.          
18.        <dependency>  
19.            <groupId>com.rabbitmqgroupId>  
20.            <artifactId>amqp-clientartifactId>  
21.            <version>5.8.0version>  
22.        dependency>  
23.          
24.        <dependency>  
25.            <groupId>commons-iogroupId>  
26.            <artifactId>commons-ioartifactId>  
27.            <version>2.6version>  
28.        dependency>  
29.  
30.        <dependency>  
31.            <groupId>org.slf4jgroupId>  
32.            <artifactId>slf4j-simpleartifactId>  
33.            <version>1.7.29version>  
34.        dependency>  
35.    dependencies>  
36.  
37.project>  
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

helloWorld模型

生产者

1.package com.lhq.rabbitmq.helloworld;  
2.  
3.import com.rabbitmq.client.Channel;  
4.import com.rabbitmq.client.Connection;  
5.import com.rabbitmq.client.ConnectionFactory;  
6.  
7.import java.io.IOException;  
8.import java.util.concurrent.TimeoutException;  
9.  
10./*HelloWord模型 生产者代码*/  
11.public class Producer {  
12.  
13.    public static final String QUEUE_NAME = "hello lhq"; // 队列名称  
14.  
15.    public static void main(String[] args) throws IOException, TimeoutException {  
16.  
17.        ConnectionFactory factory = new ConnectionFactory();  
18.        factory.setHost("10.20.20.170");           // 设置MQ所在机器IP进行连接  
19.        factory.setPort(5672);                        // 指定MQ服务端口  
20.        factory.setUsername("rabbitmq");                 // 指定MQ账号名  
21.        factory.setPassword("rabbitmq");                   // 指定MQ密码  
22.  
23.        Connection connection = factory.newConnection();    // 创建连接  
24.        Channel channel = connection.createChannel();       // 创建信道  
25.  
26.        /*  队列设置（创建队列) 
27.         *参数1：队列名称，名称不存在就自动创建 
28.         *参数2：定义队列是否持久化（重启MQ后是队列否存在）,true开启，false关闭 
29.         *参数3：exclusive 是否独占队列（设置是否只能有一个消费者使用），true独占，false非独占 
30.         *参数4：autoelete 是否在消费完成后是否自动删除队列 ，true删除,false不删除 
31.         *参数5：额外附加参数 
32.         */  
33.        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
34.  
35.        String message = "Hello RabbitMQ---LHQ";            // 需要发送的消息  
36.  
37.        /*  交换机&队列设置（指定消息使用的交换机和队列） 
38.         * 参数1： exchange交换机名称（简单队列无交换机，这里不写） 
39.         * 参数2： 有交换机就是路由key。没有交换机就是队列名称，意为往该队列里存放消息 
40.         * 参数3： 传递消息的额外设置 (设置消息是否持久化）  MessageProperties.PERSISTENT_TEXT_PLAIN设置消息持久化 
41.         * 参数4： 消息具体内容（要为 Byte类型） 
42.         */  
43.        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());  
44.  
45.        /*关闭资源*/  
46.        channel.close();  
47.        connection.close();  
48.  
49.        System.out.println("消息生产完毕");  
50.  
51.    }  
52.  
53.}  
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

消费者

1.package com.lhq.rabbitmq.helloworld;  
2.  
3.import com.rabbitmq.client.*;  
4.  
5.import java.io.IOException;  
6.import java.util.concurrent.TimeoutException;  
7.  
8./*HelloWord模型 消费者案例*/  
9.public class Consumer {  
10.  
11.    public static final String QUEUE_NAME = "hello lhq";  // 队列名称  
12.  
13.    public static void main(String[] args) throws IOException, TimeoutException {  
14.  
15.        ConnectionFactory factory = new ConnectionFactory();  
16.        factory.setHost("10.20.20.170");// 设置MQ所在机器IP进行连接  
17.        factory.setPort(5672);// 指定MQ服务端口  
18.        factory.setUsername("rabbitmq");// 指定MQ账号名  
19.        factory.setPassword("rabbitmq");// 指定MQ密码  
20.  
21.        Connection connection = factory.newConnection();// 创建连接  
22.        Channel channel = connection.createChannel();// 创建信道  
23.  
24.        /*消费者成功消费时的回调接口，这里为打印获取到的消息*/  
25.        DeliverCallback deliverCallback = (consumerTag, message) -> {  
26.            System.out.println(new String(message.getBody()));  
27.        };  
28.  
29.        /*消费者取消消费的回调*/  
30.        CancelCallback callback = consumerTag -> {  
31.            System.out.println("消息者取消消费接口回调逻辑");  
32.        };  
33.  
34.        /*  消费消息 
35.         * 参数1 ： 消费队列的名称 
36.         * 参数2 ： 消息的自动确认机制(一获得消息就通知 MQ 消息已被消费)  true打开，false关闭 (接收到消息并消费后也不通知 MQ ，常用) 
37.         * 参数3 ： 消费者成功消费时的回调接口 
38.         * 参数4 ： 消费者取消消费的回调 
39.         */  
40.        channel.basicConsume(QUEUE_NAME, true, deliverCallback, callback);  
41.  
42.        System.out.println("消费者执行完毕");  
43.  
44.    }  
45.  
46.}  
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

运行效果

WorkQueue模型

工具类抽取

RabbitMqUtil.java

1.package com.lhq.rabbitmq.utils;  
2.  
3.import com.rabbitmq.client.Channel;  
4.import com.rabbitmq.client.Connection;  
5.import com.rabbitmq.client.ConnectionFactory;  
6.  
7.public class RabbitMqUtils {  
8.  
9.    public static ConnectionFactory connectionFactory; // 放到静态代码块中  
10.  
11.    static {  
12.        connectionFactory = new ConnectionFactory();  
13.        connectionFactory.setHost("10.20.20.170");// 设置MQ所在机器IP进行连接  
14.        connectionFactory.setPort(5672);// 指定MQ服务端口  
15.        connectionFactory.setUsername("rabbitmq");// 指定MQ账号名  
16.        connectionFactory.setPassword("rabbitmq");// 指定MQ密码  
17.    }  
18.  
19.    //创建连接的方法  
20.    public static Connection getConnection() {  
21.        try{  
22.            return connectionFactory.newConnection();  
23.        } catch (Exception e) {  
24.            e.printStackTrace();  
25.        }  
26.        return null;  
27.    }  
28.  
29.    //关闭channel和connection的方法  
30.    public static void closeConnectionAndChannel(Channel channel, Connection connection) {  
31.        try {  
32.            if(channel != null) {  
33.                channel.close();  
34.            }  
35.            if (connection != null) {  
36.                connection.close();  
37.            }  
38.        } catch (Exception e) {  
39.            e.printStackTrace();  
40.        }  
41.    }  
42.}  
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

生产者

1.package com.lhq.rabbitmq.workqueue;  
2.  
3.import com.lhq.rabbitmq.utils.RabbitMqUtils;  
4.import com.rabbitmq.client.Channel;  
5.import com.rabbitmq.client.Connection;  
6.  
7.import java.io.IOException;  
8.import java.util.Scanner;  
9.  
10./** 
11. * 生产者代码 
12. */  
13.public class Task01 {  
14.  
15.    public static final String QUEUE_NAME = "hello lhq";  
16.  
17.    public static void main(String[] args) throws IOException {  
18.  
19.        Connection connection = RabbitMqUtils.getConnection();  
20.        Channel channel = connection.createChannel();  
21.  
22.        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
23.  
24.        Scanner scanner = new Scanner(System.in);  
25.        while (scanner.hasNext()) {  
26.            String message = scanner.next();// 从控制台获取消息  
27.  
28.            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());  
29.            System.out.println("消息发送完毕" + message);  
30.        }  
31.  
32.    }  
33.}  

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

消费者

1.package com.lhq.rabbitmq.workqueue;  
2.  
3.import com.lhq.rabbitmq.utils.RabbitMqUtils;  
4.import com.rabbitmq.client.*;  
5.  
6.import java.io.IOException;  
7.  
8./** 
9. * 工作进程 
10. */  
11.public class Work01 {  
12.  
13.    public static final String QUEUE_NAME = "hello lhq";  
14.  
15.    public static void main(String[] args) throws IOException {  
16.  
17.        Connection connection = RabbitMqUtils.getConnection();  
18.        Channel channel = connection.createChannel();  
19.  
20.        DeliverCallback deliverCallback = (consumerTag, message) -> {  
21.            System.out.println("接收到的消息：" + new String(message.getBody()));  
22.        };  
23.  
24.        CancelCallback cancelCallback = consumerTag -> {  
25.            System.out.println(consumerTag + "消费者取消消费接口回调逻辑");  
26.        };  
27.  
28.        System.out.println("消费者B等待接收消息---------");  
29.  
30.        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);  
31.    }  
32.  
33.}  

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

运行效果

消息应答

概念

为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，消息应答就是:消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。

自动应答

消息发送后立马就被认为是发送成功了，这时候如果消费者在接收之前出现连接关闭或者channel关闭，那么消息就丢失了；当然消费者也可以传递过载的消息，但是如果消费者接收过多导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。

消息应答方式

A.Channel.basicAck(用于肯定确认)
RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了

B.Channel.basicNack(用于否定确认)

C.Channel.basicReject(用于否定确认)
与 Channel.basicNack 相比少一个参数。不处理该消息了直接拒绝，可以将其丢弃了

Multiple

手动应答的好处是可以批量应答并且减少网络拥堵；
坏处就是会导致消息的丢失；

消息自动重新入队

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

确认应答代码

1./*消费者成功回调逻辑*/  
2.DeliverCallback deliverCallback = (consumerTag, delivery) -> {  
3.    String message = new String(delivery.getBody());  
4.    SleepUtils.sleep(30);  
5.    System.out.println("接收到消息" + message);  
6.  
7.    /** 
8.     * 1. 消息标记tag 
9.     * 2. 是否批次消费消息（true为应答该队列中所有消息，false为只应答接受到的消息） 
10.     */  
11.    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);  
12.};  
1
2
3
4
5
6
7
8
9
10
11
12

持久化

队列持久化

RabbitMQ突发崩溃的时候，如果队列没有做持久化将会导致队列和队列内消息丢失。在队列创建时指定第二个参数为true即可。如果队列没有被持久化，在RabbitMQ重启后将被销毁。

1.// 队列持久化 durable  
2.boolean durable = true;  
3./** 
4. * 生成一个队列 
5. * 1.队列名称 
6. * 2.队列里面的消息是否持久化 默认消息存储在内存中 ture 开启 false关闭 
7. * 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费 
8. * 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除 
9. * 5.其他参数 
10. */  
channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null);  
1
2
3
4
5
6
7
8
9
10
11

消息持久化

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉RabbitMQ将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强。最好的方式为**MQ集群+发布确认+消息存入Redis(开启AOF)**保证100%不丢失。

1.// 消息持久化 basicProperties:MessageProperties.PERSISTENT_TEXT_PLAIN  
2.channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));  
1
2

不公平分发

轮训分发的时候，比方说有两个消费者在处理任务，其中有个消费者 1 处理任务的速度非常快，而另外一个消费者 2 处理速度却很慢，这个时候如果还是采用轮训分发就会使得处理速度快的这个消费者很大一部分时间处于空闲状态，而处理慢的那个消费者一直在干活，这种分配方式在这种情况下其实就不太好，但是 RabbitMQ 并不知道这种情况它依然很公平的进行分发。 为了避免这种情况，我们可以设置参数 channel.basicQos(1);

1.channel.basicQos(1); // 设置不公平分发 1则代表不公平分发 一个队列的所有消费者都需要设置  
1

预取值

预取值就是该值定义通道上允许的未确认消息的最大数量。消息应答和Qos预取值对用户吞吐量有着很大影响。如果Qos取的过于高，，那么积压在信道中未处理的消息也会变得多，从而增加了消费者的RAM消耗；如果过于低了，那么吞吐量就会变得很低。正常在100-300间最佳。

1./** 
2. * prefetchCount:默认为 0，采用轮询规则 
3. * 设置为 1时，采用不公平分发（能者多劳） 
4. * 设置为 >1 时，代表预取值（当前消费者堆积未处理消息的缓冲区大小） 
5. */  
6.int prefetchCount = 10;  
7.channel.basicQos(prefetchCount);  
1
2
3
4
5
6
7

发布确认

开启发布确认

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法。

1.channel.confirmSelect(); //  设置消息确认  
1

单个消息发布确认

它是一种同步确认发布的方式，这种确认方式有一个最大的缺点就是:发布速度特别的慢。

生产者代码

1./* 发布确认 单个确认 */  
2.public class SingleConfirms {  
3.  
4.    public static final int MESSAGE_COUNT = 1000;  
5.    public static final String QUEUE_NAME = "single confirm"; // 随机生成一个队列名字  
6.  
7.    public static void main(String[] args) throws Exception {  
8.  
9.        Connection connection = RabbitMqUtils.getConnection();  
10.        Channel channel = connection.createChannel();  
11.  
12.        channel.queueDeclare(QUEUE_NAME, true, false, false, null);  
13.        channel.confirmSelect(); //  设置消息确认  
14.        long startTime = System.currentTimeMillis();  //  设置开始时间  
15.        for (int i = 0; i < MESSAGE_COUNT; i++) {  
16.  
17.            String message = " 当前是 " + i + " 条消息 ";  
18.            // 消息发送  
19.            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));  
20.  
21.            boolean flag = channel.waitForConfirms();  //  等待消息发布确认通知 参数为最大等待时间  
22.            if ( flag ) {  
23.                System.out.println( "消息" + i + "发送成功");  
24.            }  
25.        }  
26.  
27.        long endTime = System.nanoTime();  // 结束时间  
28.        System.out.println("单个确认模式耗时：" + (endTime - startTime));  
29.    }  
30.  
31.}  
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

批量确认发布

与单个确认发布相比，极大提高吞吐量；但是当发生故障时不好排查具体是哪一条消息报错。

生产者代码

1./* 批量确认发布 */  
2.public class BatchConfirms {  
3.    public static final int MESSAGE_COUNT = 1000;  
4.    public static final String QUEUE_NAME = "batch_confirms";  
5.  
6.    public static void main(String[] args) throws Exception{  
7.        Connection connection = RabbitMqUtils.getConnection();  
8.        Channel channel = connection.createChannel();  
9.  
10.        channel.confirmSelect(); // 开启发布确认  
11.        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
12.  
13.        long startTime = System.currentTimeMillis();  //  设置开始时间  
14.  
15.        for (int i = 0; i < MESSAGE_COUNT; i++) {  
16.            String message = "当前是：" + i + "条消息";  
17.            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));  
18.  
19.            if (i % 100 == 0) {  
20.                boolean flag = channel.waitForConfirms();  
21.                if (flag) {  
22.                    System.out.println("消息" + i + "发送成功");  
23.                }  
24.            }  
25.        }  
26.  
27.        long endTime = System.currentTimeMillis(); // 设置结束时间  
28.        System.out.println("批量确认模式耗时：" + (endTime - startTime));  
29.    }  
30.}  
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

异步确认发布

通过回调函数确认

package com.lhq.rabbitmq.ConfirmSelect;

import com.lhq.rabbitmq.Utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.Connection;
import com.sun.javaws.IconUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/* 异步确认 */
public class PublishMessageAsync {

    public static final int MESSAGE_COUNT = 100;
    public static final String QUEUE_NAME = "message async";

    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMqUtils.getConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目 只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        /**
        * 确认收到消息的一个回调
        * 1.消息序列号
        * 2.true 可以确认小于等于当前序列号的消息
        * false 确认当前序列号消息
        */
        ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
            if (multiple) {
                //返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                //清除该部分未确认消息
                confirmed.clear();
            }else{
                //只清除当前序列号的消息
                outstandingConfirms.remove(sequenceNumber);
            }
        };

        /**
         * 消息未被确认的一个回调
         */
        ConfirmCallback nackCallack = (sequenceNumber, multiple) -> {
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("发布的消息" + message + "未被确认，" + "序列号是：" + sequenceNumber);
        };
        /**
         * 添加一个异步确认的监听器
         * 1. 成功的回调
         * 2. 失败的回调
         */
        channel.addConfirmListener(ackCallback, null);
        long beginTime = System.currentTimeMillis();
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
            channel.basicPublish("", QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
        }
        long endTime = System.currentTimeMillis();
        System.out.println("发布" + MESSAGE_COUNT + "个异步确认消息,耗时" + (endTime - beginTime) + "ms");
    }
}

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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76