Java中动态声明与绑定Rabbit MQ队列以及延迟队列的实现与使用

Posted 2022-11-09 YQS_Love

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Java中动态声明与绑定Rabbit MQ队列以及延迟队列的实现与使用相关的知识，希望对你有一定的参考价值。

一概述

通常，我们会通过Spring schemal来配置队列基础设施、队列声明以及绑定等功能，这样让我们能够很方便的通过Spring注入对象去使用它，但有不足之处就是，在项目中如果我们想要多个队列去分担不同的任务，此时我们不得不创建很多不同的Rabbit MQ Spring schemal，那么这种做法就显得太过繁琐与笨重了。反之，在Java代码里动态的去声明和绑定队列，就会方便很多了，而在schemal中我们只需引入Rabbit MQ的相关配置即可。本篇博客会讲解如何在Java代码中动态的声明与绑定队列以及延迟队列的实现。

注意：

本篇博客介绍的是Java语言下的使用，客户端使用的是Spring AMQP，版本为1.7.7(详情见pom.xml)；
本篇博客不使用Rabbit MQ的数据对象转化，如有需要须自行实现；
代码中声明的Exchange都为D型的，如果需要别的类型，可自行抽取代码。

二配置Rabbit MQ

pom.xm 引入Spring AMQP

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>1.7.7.RELEASE</version>
</dependency>

在 config.properties（一个自定义的属性配置文件）中配置Rabbit MQ相关，需要在Spring的schemal 中导入

# Rabbit MQ
rabbit.username=test
rabbit.password=123123
rabbit.port=5672
rabbit.host=192.168.30.218
rabbit.virtual.host=/rabbit

编写applicationContext-rabbitmq.xml schemal

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
           http://www.springframework.org/schema/rabbit/spring-rabbit.xsd
           http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans.xsd">

    <rabbit:connection-factory id="rabbitMqConnectionFactory"
                               host="$rabbit.host" port="$rabbit.port"
                               username="$rabbit.username"
                               password="$rabbit.password"
                               virtual-host="$rabbit.virtual.host"
                               channel-cache-size="300"
                               publisher-confirms="true"/>

    <rabbit:template id="rabbitAmqpTemplate" connection-factory="rabbitMqConnectionFactory"/>

    <rabbit:admin id="rabbitAdmin" connection-factory="rabbitMqConnectionFactory"/>

</beans>

抽取生产者和消费者公共配置接口IRabbitMqConfig

package com.bell.rabbitmq;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 18:33
 * Copyright © Bell All Rights Reserved.
 */
public interface IRabbitMqConfig 

    /**
     * queue name
     *
     * @return
     */
    String queueName();

    /**
     * queue exchange name
     *
     * @return
     */
    String queueExchangeName();

    /**
     * queue route key
     *
     * @return
     */
    String queueRouteKey();

抽取生产者和消费者公共配置抽象类AbstractRabbitMqBase并实现IRabbitMqConfig接口，但在抽象类型不实现IRabbitMqConfig接口

package com.bell.rabbitmq;

import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 18:37
 * Copyright © Bell All Rights Reserved.
 */
public abstract class AbstractRabbitMqBase implements IRabbitMqConfig 

    @Resource
    private RabbitAdmin rabbitAdmin;

    @Resource
    private RabbitTemplate rabbitAmqpTemplate;

    @PostConstruct
    private void init() 

        Queue queue = new Queue(queueName());
        DirectExchange exchange = new DirectExchange(queueExchangeName());

        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareExchange(exchange);
        rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(queueRouteKey()));

    

    /**
     * 发布字符串信息到队列中
     *
     * @param
     */
    protected void publishMessage(String message) 
        rabbitAmqpTemplate.convertAndSend(queueExchangeName(), queueRouteKey(), message);
    

    /**
     * 发布Message对象信息到队列中
     *
     * @param message
     */
    protected void publishMessage(Message message) 
        rabbitAmqpTemplate.send(queueExchangeName(), queueRouteKey(), message);

抽取消费者公共抽象类并继承AbstractRabbitMqBase，同时实现ChannelAwareMessageListener，此处我们使用ChannelAwareMessageListener去接收消息，除此之外，我们需要在增加一个抽象方法getConsumerCount()，用于配置要启用多少个消费者,同时需要实现onDestroy()方法，在类销毁时去断开与MQ服务器的链接，而不是异常退出，保证消息不丢失或被正常ACK

package com.bell.rabbitmq;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.Objects;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 18:50
 * Copyright © Bell All Rights Reserved.
 */
public abstract class AbstractRabbitMqConsumer extends AbstractRabbitMqBase implements ChannelAwareMessageListener 

    @Resource
    private ConfigService configService;

    @Resource
    private ConnectionFactory consumerConnectionFactory;

    private SimpleMessageListenerContainer[] rabbitMqListener = null;

    @PostConstruct
    private void init() 
    
        rabbitMqListener = new SimpleMessageListenerContainer[getConsumerCount()];

        for (int i = 0; i < getConsumerCount(); ++i) 
            rabbitMqListener[i] = new SimpleMessageListenerContainer(consumerConnectionFactory);
            rabbitMqListener[i].setMessageListener(this);
            rabbitMqListener[i].setAcknowledgeMode(AcknowledgeMode.MANUAL);
            rabbitMqListener[i].setQueueNames(queueName());
            rabbitMqListener[i].start();
        
    

    /**
     * 需要创建多少个消费者
     *
     * @return
     */
    protected abstract int getConsumerCount();

    @PreDestroy
    private void onDestroy() 

        if (Objects.isNull(rabbitMqListener) || rabbitMqListener.length <= 0) 
            return;
        

        for (int i = 0; i < getConsumerCount(); ++i) 
            rabbitMqListener[i].destroy();

三使用Rabbit MQ

完成了Rabbit MQ的基本引入与配置后，就可以去使用它了

生产者的使用。定义一个生产者类TestRabbitMqProducer继承AbstractRabbitMqBase,然后在需要使用的地方通过Spring注入此类，就可以发送消息到队列里了

package com.bell.rabbitmq.test;

import org.springframework.stereotype.Service;
import bell.util.JSONUtil;

import java.util.Objects;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 16:25
 * Copyright © Bell All Rights Reserved.
 */
@Service
public class TestRabbitMqProducer extends AbstractRabbitMqBase 

    public Boolean publish(String data) 

        // data 字符串我使用json格式的，这样方便反序列化，当然可以使用Rabbit MQ的convert，由于篇幅有限，不做介绍，请自行实现
        this.publishMessage(data);

        return true;
    

    @Override
    public String queueName() 
        return "rabbitmq.test.queue";
    

    @Override
    public String queueExchangeName() 
        return "rabbitmq.test.exchange";
    

    @Override
    public String queueRouteKey() 
        return "rabbitmq.test.route.key";
    

    // 对于 queue,exchange,route key等参数可以放到一个常量类中，一处定义，多处可用，还能保证生产者与消费者不一致

消费者的使用。创建消费者类TestRabbitMqConsumer继承AbstractRabbitMqConsumer类，并实现所有的方法才可使用，此类只要使用@Service标注后，就可开始消费，不用做其他的操作

package com.bell.rabbitmq.test;

import org.springframework.stereotype.Service;
import bell.util.JSONUtil;

import java.util.Objects;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 16:25
 * Copyright © Bell All Rights Reserved.
 */
@Service
public class TestRabbitMqConsumer extends AbstractRabbitMqConsumer 


    /**
     *   onMessage方法中不要抛出异常，否则会阻塞此消费者，导致服务端不在向此消费者推送消息
     */  
    @Override
    public void onMessage(Message message, Channel channel) throws IOException 

        // 我们发送的消息类型是已bytes数组的形式存在的
        String eventMessage = new String(message.getBody());

        System.out.println(eventMessage);

        // 确认收到消息
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        // 确认收到消息，但是业务异常了，需要重回队列
        // channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            
    

    @Override
    protected int getConsumerCount() 
        return 1;
    

    @Override
    public String queueName() 
        return "rabbitmq.test.queue";
    

    @Override
    public String queueExchangeName() 
        return "rabbitmq.test.exchange";
    

    @Override
    public String queueRouteKey() 
        return "rabbitmq.test.route.key";
    

    // 对于 queue,exchange,route key等参数可以放到一个常量类中，一处定义，多处可用，还能保证生产者与消费者不一致

到这里，动态的声明和绑定队列就完了，接下来，我将继续讲解如何配置与使用延迟队列。

四延迟队列的配置与使用须知

延迟队列的实现有两种方式，第一种是使用插件的方式（详情见官网），第一种方式是通过插件的方式去启用延迟队列，由于插件是官方Rabbit MQ不自带的，所以在Rabbit MQ的后台管理中心是看不到延迟队列信息的，不太利于观察和维护，但使用与否，取决于个人。第二种方式是使用两个队列去实现，一个队列充当消息计时队列，当这些消息变成死信之后，如果不配置死信转发机制，那么这些死信将会被丢弃，反之，要想这些消息被消费，就需要另一个队列接收这些死信，从而让它在次被消费，从而实现延迟的功能。使用本篇博客所配置的延迟队列需要注意以下事项：

1）不对队列设置TTL。

2）不对单个消息设置TTL。也就是说所有的消息得过期时间都是一样的，过期从进入队列开始计算，因消息到达队列有先后顺序，如果同一个队列中的每个消息时间都不一样，那么队头的消息时间还没有过期，而队列中间的消息过期了，队列中间的消息不会立马被转发，只有当他到达队头后才能被转发，因此，为保证过期消息的过期时间不远大于设定的时间，本博客讲解的配置不对队列和单个消息设置不同的时间。

3）因采用的是使用两个队列实现延迟队列，因此需要结合本篇博客前部分的配置，请熟知。

4）延迟队列不得有消费，否则就无法实现延迟功能。

五延迟队列的配置与使用

队列配置参数解释

1）参数x-dead-letter-exchange为死信转发Exchange;

2）参数x-dead-letter-routing-key为死信转发Route key;

3）参数x-message-ttl为消息在队列里的生存时间；

创建延迟队列配置抽象类AbstractRabbitDelayMqBase并实现IRabbitMqConfig接口，接口在抽象类中不实现。同时创建messageTtl()、deadLetterRoutingKey()、deadLetterExchange()三个抽象方法，方法作用描述见代码

package com.bell.rabbitmq;

import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 18:37
 * Copyright © Bell All Rights Reserved.
 */
public abstract class AbstractRabbitDelayMqBase implements IRabbitMqConfig 

    @Resource
    private RabbitAdmin producerRabbitAdmin;

    @Resource
    private RabbitTemplate producerAmqpTemplate;

    @PostConstruct
    private void init() 

		// 此处配置延迟队列相关参数
        Map<String, Object> args = new HashMap<>(3);
        args.put("x-dead-letter-exchange", deadLetterExchange());
        args.put("x-dead-letter-routing-key", deadLetterRoutingKey());
        args.put("x-message-ttl", messageTtl());

        DirectExchange exchange = new DirectExchange(queueExchangeName());
        Queue queue = new Queue(queueName(), true, false, false, args);

        producerRabbitAdmin.declareQueue(queue);
        producerRabbitAdmin.declareExchange(exchange);
        producerRabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(queueRouteKey()));
    

    /**
     * 消息生存时间 (单位毫秒)
     *
     * @return
     */
    protected abstract Integer messageTtl();

    /**
     * 死信交Routing
     *
     * @return
     */
    protected abstract String deadLetterRoutingKey();

    /**
     * 死信Exchange
     *
     * @return
     */
    protected abstract String deadLetterExchange();

    /**
     * 发布消息到队列中
     *
     * @param
     */
    protected void publishMessage(String message) 
        producerAmqpTemplate.convertAndSend(queueExchangeName(), queueRouteKey(), message);
    

    /**
     * 发布消息队列中
     *
     * @param message
     */
    protected void publishMessage(Message message) 
        producerAmqpTemplate.send(queueExchangeName(), queueRouteKey(), message);

创建延迟队列生产者类TestRabbitMqDelayProducer继承AbstractRabbitDelayMqBase类，并实现所有方法，在实现方法时需要注意，延迟队列死信转发的Exchange和Route key必须同接收死信队列的Exchange和Route key保持一致，且延迟队列不需要消费者，接收死信的队列的生产者可根据业务需求可有可无

package com.bell.rabbitmq.test;

import org.springframework.amqp.core.MessageProperties;
import org.springframework.stereotype.Service;
import bell.util.JSONUtil;

import javax.annotation.PostConstruct;
import java.util.Objects;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 14:53
 * Copyright © Bell All Rights Reserved.
 */
@Service
public class TestRabbitMqDelayProducer extends AbstractRabbitDelayMqBase 

    /**
     * 消息过期时间 (一天)
     */
    private static final int MESSAGE_DELAY_TIME = 24 * 3600 * 1000;

    public Boolean publishMessage(String data) 

        // data为json字符串，同样，可以使用Rabbit MQ的convert,自行实现
        publishMessage(data);
        
        return true;
    

    @Override
    protected Integer messageTtl() 
        return MESSAGE_DELAY_TIME;
    


    @Override
    public String queueName() 
        return "rabbitmq.test.delay.queue";
    

    @Override
    public String queueExchangeName() 
        return "rabbitmq.test.delay.exchange";
    

    @Override
    public String queueRouteKey() 
        return "rabbitmq.test.delay.route.key";
    

    @Override
    protected String deadLetterRoutingKey() 
        return "rabbitmq.test.delay.receive.route.key";
    

    @Override
    protected String deadLetterExchange() 
        return "rabbitmq.test.delay.receive.exchange";
    

    // 对于 queue,exchange,route key等参数可以放到一个常量类中，一处定义，多处可用，还能保证生产者与消费者不一致

创建延迟队列消费者类TestRabbitMqDelayConsumer并继承AbstractRabbitMqConsumer抽象类

package com.bell.rabbitmq.test;

import org.springframework.stereotype.Service;
import bell.util.JSONUtil;

import java.util.Objects;

/**
 * @Author: yqs
 * @Date: 2019/1/25
 * @Time: 16:25
 * Copyright © Bell All Rights Reserved.
 */
@Service
public class TestRabbitMqDelayConsumer extends AbstractRabbitMqConsumer 


    /**
     *   onMessage方法中不要抛出异常，否则会阻塞此消费者，导致服务端不在向此消费者推送消息
     */  
    @Override
    public void onMessage(Message message, Channel channel) throws IOException 

        // 我们发送的消息类型是已bytes数组的形式存在的
        String eventMessage = new String(message.getBody());

        System.out.println(eventMessage);

        // 确认收到消息
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        // 确认收到消息，但是业务异常了，需要重回队列
        // channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            
    

    @Override
    protected int getConsumerCount() 
        return 1;
    

    @Override
    public String queueName() 
        return "rabbitmq.test.delay.receive.queue";
    

    // 此处的 Exchange 和 Route key必须同延迟队列里声明的死信转发保持一致

    @Override
    以上是关于Java中动态声明与绑定Rabbit MQ队列以及延迟队列的实现与使用的主要内容，如果未能解决你的问题，请参考以下文章 
 rabbit mq 基础流程（转）
 Spring Boot：使用Rabbit MQ消息队列
 Rabbit MQ一些参数解释
 Rabbit MQ部署步骤
 何时使用 RabbitMQ 声明/绑定队列和交换
 具有单队列的 Rabbit MQ 中的多生产者、多消费者

Java中动态声明与绑定Rabbit MQ队列以及延迟队列的实现与使用

一 概述

二 配置Rabbit MQ

三 使用Rabbit MQ

四 延迟队列的配置与使用须知

五 延迟队列的配置与使用

一概述

二配置Rabbit MQ

三使用Rabbit MQ

四延迟队列的配置与使用须知

五延迟队列的配置与使用