RabbitMQ------其他知识点（幂等性、优先级队列、惰性队列）（九）

RabbitMQ------其他知识点（）（九）

幂等性

用户对于统一操作发起的一次请求或者多次请求的结果是一致的，不会因为多次点击而产生副作用。同一次还款，通过多次点击不会生成多条还款记录，一个人的注册按钮，多次点击不会生成多条用户信息。

解决思路

MQ消费者的幂等性的解决一般使用全局ID，或者写一个唯一标识比如时间戳，或者UUID或者订单消费者MQ中的消息，也可以利用MQ的该id来判断，或者可按自己的规则生成一个全局唯一id，每次消费消息时用该id先判断该消息是否已消费过。

消费端的幂等性保障

业界主流的幂等性有两种操作：1.唯一ID+指纹码机制，利用数据库主键去重， 2.利用redis的原子性去实现。

唯一ID+指纹码机制

指纹码：基本上都是由业务规则拼接而成，但是一定要保证唯一性。然后利用查询语句进行判断这个id是否存在数据库中。
优势：实现简单，一个拼接加上查询判断是否重复。
劣势：高并发时，单个数据库有写入性能瓶颈，当然也可以采用分库分表提升性能。

Redsi原子性（建议使用）

利用Redis执行setnx命令（只有不存在时候才设置），具有天然幂等性，从而实现不重复消费

优先级队列

对于同样的消息，有优先级处理，优先级高的，优先处理。如果发现是一个大客户的小新，就给与比较高的优先级，否则采用默认优先级。
队列的先进先出。
消费者在获取优先级队列消息时，是先把消息进行排队，排完队再获取。
优先级队列的优先级是0-255，却大，越优先被消费。

页面配置

在管理页面中，点击Queues—》Add a new queue-----》Maxium priority
点击后，会在Arguments中，自动带入参数 x-max-pripority
可以在value，中输入0-255之间的数字，但是建议10，数字越大，排序时开销也越大。

代码配置

1.首先在队列的代码中添加优先级

parms.put("x-max-priority", 10)
channel.queueDeclare("hello",true,false,false,parms);
1
2

2.发消息时，设置消息的优先级。
优先级别在队列优先级的范围内。

new AMQP.BasicProperties().builder().priority(5).builder();
1

只有队列和消息同时设置优先级，才能够使用该功能
且 消费者需要等待消息已经发送到队列中才能够去消费，应为此时才有机会对消息进行排序。
【将需要排序的消息，全部，发送到队列中后，消费者再消费】

示例代码：
配置类中新增优先级队列

    //优先级队列
    public static final String PRIPORITY_QUEUE_NAME = "pripority_queue";
    //声明 优先级队列
    @Bean("priorityQueue")
    public Queue priporityQueue(){
        //创建队列的两种方式
        //QueueBuilder.durable(CONFIRM_QUEUE_NAME).build();
        HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-max-priority",10);
        return  new Queue(PRIPORITY_QUEUE_NAME,true,false,false,arguments);
    }
1
2
3
4
5
6
7
8
9
10
11

生产者：

    //开始发消息
    @GetMapping("/sendExpirationMessage/{message}")
    public void sendPriMsg(@PathVariable String message){
        log.info("当前时间:{},发送一条消息给优先级队列：{},优先级为：{}ms",new Date().toString(),message);

        /**
         * 交换机
         * routingkey
         * message
         * Priority,可以设置存活时间
         */
        for (int i = 0; i < 10; i++) {
            int finalI = i;
            MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setPriority(finalI);
                    return message;
                }
            };
            rabbitTemplate.convertAndSend("X","XC","优先级的队列:"+message,messagePostProcessor);
        }
        
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

结论：可以实现优先级高的消息被优先消费。
但是也需要实现消息堆积
1.自己模拟的话，只能先启动生产者发消息，再启动消费者进行消费。
按这个思路生产上，应该是…
通过定时任务的方式开启消费者。

2.或者消费者一直启动，能否通过配置或者代码的方式，实现RabbitMQ对消息进行堆积操作？遗留问题

惰性队列

惰性队列会尽可能的将消息存入磁盘中，在消费者消费到相应的消息时，才会被加载到内存中。
使用场景：消费者下线了，宕机了或者由于维护而关闭了，导致长时间内不能消费消息时，造成堆积时，惰性队列就很有必要了。

两种模式

队列具有两种模式：default和lazy，默认的为default模式，lazy模式就是惰性队列的模式，可以通过调用channel.queueDeclare方法的时候再参数中设置。也可以通过Policy的方式设置。
两种方式都设置的话，Policy的方式具有更高的优先级。
如果通过声明的方式改变已有队列的模式的话，只能先删除队列，再重新声明一个新的。
声明方式设置：

HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-queue-mode","lazy");
        return  new Queue(PRIPORITY_QUEUE_NAME,true,false,false,arguments);
或者
channel.queueDeclare("myqueue",false,false,false,arguments);
1
2
3
4
5

页面设置：
与页面设置优先级队列一样，选择Lazy mode，会自动将参数点出来。

结论：在发送1百万条消息时，每条消息大概占1KB的情况下，普通队列占用内存时1.2GB，而惰性队列仅仅占用1.5MB（因为只放在磁盘中，内存中仅存放索引）