一、为什么要封装

        在软件项目开发中，基本上都是多人共同开发，尤其是大型项目。而消息队列更是大多数业务都会使用，前期基本上都是在业务中直接注入RabbitTemplate，创建各自的交换机、队列、监听者完成业务开发，但是到了项目后期，大量的交换机和队列的定义代码充斥在各个模块，每次需要回溯业务逻辑的时候都在不停的翻找代码，寻找监听者。

致命的是，通常我们找到了发消息的代码，由于通过RabbitTemplate发送消息是直接发给交换机的，想要找到处理消息的消费者业务代码，还需要通过交换机定义代码——交换机队列绑定——队列——监听者，这样繁琐的操作。同样的当你找了消息消费者的代码却不知道都有项目哪块对该队列发送了消息，如此总总浪费了开发人员大量的时间。

于是笔者为了消灭这种情况决定对RabbitMQ操作进行封装，目的是实现以下功能：

1、代码统一，交换机和队列复用，消灭项目定义交换机及队列的代码。

2、使用简单，像使用工具类一样，拿来即用。

3、流程清晰，在发送消息的地方指明消费者代码位置，同时通过消费者代码亦可找到有多少消息发送者。

4、异常处理增强，当消费者报错时候，通过优雅的方式处理异常，而不是rabbitmq反复重试，导致满屏错误日志。

二、代码

封装后总共有4个.java文件，分别是：

MqConf.java——>用于配置RabbitMq，定义公共交换机、队列、统一消费者等。

MqListener<T>.java——>抽象类，使用时定义消息消费者业务代码需要继承此类实现onReceive和onError方法。

MqTool.java——>定义对外调用方法，实际使用时直接调用该类方法发送消息。

MqMsg<T>.java——>实际发送的消息包装体。

MqConf.java

import com.bohuikeji.frame.module.admin.base.MqListener;
import com.bohuikeji.frame.module.admin.entity.mq.MqMsg;
import com.bohuikeji.frame.module.admin.thread.CachedThreadPool;
import com.bohuikeji.frame.module.admin.thread.MqListenerThread;
import com.bohuikeji.frame.module.admin.utils.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMQ配置
 * @author sjh
 */
@Slf4j
@Configuration("ranch-rabbitmq-conf")
public class MqConf {

    /**
     * 点对点消息队列
     * 默认交换机
     */
    public static final String QUEUE_RANCH_ONE_TO_ONE="queue.ranch.one.to.one";
    /**
     * 发送一对多(广播)消息队列
     */
    public static final String QUEUE_RANCH_ONE_TO_ALL="queue.ranch.one.to.all."+System.currentTimeMillis();

    /**
     * 延时消息队列
     */
    public static final String QUEUE_RANCH_DELAY="queue.ranch.delay";

    /**
     * 广播交换机
     */
    public static final String EXCHANGE_FANOUT_RANCH ="exchange.fanout.ranch";

    /**
     * 延时交换机
     */
    public static final String EXCHANGE_DELAY_RANCH="exchange.delay.ranch";


    @Bean
    public Queue oneToOneQueue() {return new Queue(QUEUE_RANCH_ONE_TO_ONE, true,false,true);}

    @Bean
    public Queue oneToAllQueue(){return new Queue(QUEUE_RANCH_ONE_TO_ALL, true,false,true);}

    @Bean
    public Queue delayQueue(){return new Queue(QUEUE_RANCH_DELAY, true,false,true);}

    /**
     * 广播交换机
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange(EXCHANGE_FANOUT_RANCH);
    }

    /**
     * 延时交换机
     * @return
     */
    @Bean
    public CustomExchange delayExchange(){
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(EXCHANGE_DELAY_RANCH,"x-delayed-message",true, false,args);
    }

    /**
     * 广播交换机绑定
      * @param oneToAllQueue
     * @param fanoutExchange
     * @return
     */
    @Bean
    public Binding fanoutBuild(Queue oneToAllQueue, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(oneToAllQueue).to(fanoutExchange);
    }

    /**
     * 延时交换机绑定
     * @param delayQueue
     * @param delayExchange
     * @return
     */
    @Bean
    public Binding delayBuild(Queue delayQueue, CustomExchange delayExchange){
        return BindingBuilder.bind(delayQueue).to(delayExchange).with(QUEUE_RANCH_DELAY).noargs();
    }


    /**
     * 统一监听监听
     * @param msg
     */
    @RabbitHandler
    @RabbitListener(queues = {QUEUE_RANCH_ONE_TO_ONE,QUEUE_RANCH_DELAY,"#{oneToAllQueue.name}"},priority = "${spring.rabbitmq.priority}")
    private void news(MqMsg<?> msg) {
        MqListener listener;
        try {
            listener= SpringUtil.getBean(msg.getListener());
        }catch (Exception e){
            log.error("【消息队列监听者业务异常】监听类：{}，缺少@Service注解,已经尝试手动注册。",msg.getListener(),e);
            SpringUtil.registerBean(msg.getListener());
            listener= SpringUtil.getBean(msg.getListener());
        }
        try {
            CachedThreadPool.execute(new MqListenerThread(listener,msg.getData()));
        } catch (Exception e) {
            log.error("【消息队列监听者业务异常】位于：{},错误信息：{}",listener.getClass(),e.getMessage(),e);
            listener.onError(msg.getData(),e);
        }
    }
}

MqListener.java 

/**
 * 消息监器
 * @author Admin
 */
public abstract class MqListener<T> {

    /**
     * 接收到消息会调用此方法
     * @param msg 消息体
     * @return 【保留返回值】暂无实际用途，后续可能用于支持消息失败重试
     */
    public abstract boolean onReceive(T msg);

    /**
     * 当消费者业务代码出现异常时会调用该方法
     * @param msg 消息体
     * @param e 异常信息
     */
    public abstract void onError(T msg,Exception e);

}

 MqTool.java

import com.bohuikeji.frame.module.admin.base.MqListener;
import com.bohuikeji.frame.module.admin.config.MqConf;
import com.bohuikeji.frame.module.admin.entity.mq.MqMsg;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import java.time.Duration;

/**
 * 消息队列发送工具
 * @author sjh
 */
@Component
@RequiredArgsConstructor
public class MqTool {

    final RabbitTemplate rabbitTemplate;

    /**
     * 发送普通消息
     * @param listenerClass 监听者
     * @param msg 消息体
     */
    public <T> void sendMsg(T msg,Class<? extends MqListener<T>> listenerClass){
        rabbitTemplate.convertAndSend(MqConf.QUEUE_RANCH_ONE_TO_ONE,new MqMsg<T>(msg,listenerClass));
    }

    /**
     * 发送广播消息
     * @param listenerClass 监听者
     * @param msg 消息体
     */
    public <T> void sendMsgToAll(T msg,Class<? extends MqListener<T>> listenerClass){
        rabbitTemplate.convertAndSend(MqConf.EXCHANGE_FANOUT_RANCH,"",new MqMsg<T>(msg,listenerClass));
    }

    /**
     * 发送延迟消息
     * @param msg 消息体
     * @param listenerClass 监听者
     * @param duration 延时时间段
     */
    public <T> void sendDelayMsg(T msg, Class<? extends MqListener<T>> listenerClass, Duration duration){
        rabbitTemplate.convertAndSend(MqConf.EXCHANGE_DELAY_RANCH, MqConf.QUEUE_RANCH_DELAY, new MqMsg<T>(msg,listenerClass), message ->{
            message.getMessageProperties().setHeader("x-delay",duration.toMillis());
            return message;
        });
    }
}

 MqMsg.java

import com.bohuikeji.frame.module.admin.base.MqListener;
import lombok.Data;
import java.io.Serializable;

/**
 * 消息队列消息体
 * @author sjh
 * @param <T>
 */
@Data
public class MqMsg<T> implements Serializable {

    /**
     * 消息体
     */
    T data;
    /**
     * 接收者
     */
    Class<?extends MqListener<T>> listener;

    /**
     * 构造方法
     * @param msg
     * @param listener
     */
    public MqMsg(T msg, Class<? extends MqListener<T>> listener) {
        this.data=msg;
        this.listener=listener;
    }
}

 在MqConf中统一监听者收到消息后，是用线程池给每个消息分配一个线程去处理的，这里把线程池的代码也贴出来，还用到一个springUtils的工具，用于获取每个消息对应消费者的bean，以下一同贴出代码。

import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import static java.lang.Integer.MAX_VALUE;

/**
 * 线程池工厂类
 *
 * @author sjh
 */
public class CachedThreadPool {

    /**
     * 静态内部类单例，初始话线程池实列
     */
    private static class ThreadPoolHolder {
        //定义一个线程池，静态内部类单例方式
        private static final ExecutorService EXEC;

        static {
            /**
             * 实例化缓存线程池
             * 适用于大量短生命周器类型的任务
             */
            EXEC = new ThreadPoolExecutor(4, MAX_VALUE,
                    60L, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),new CustomizableThreadFactory("ranchThread"));
        }
    }

    /**
     * 执行线程任务
     *
     * @param command 一个需要执行的线程任务
     */
    public static void execute(Runnable command) throws Exception {
        ThreadPoolHolder.EXEC.execute(command);
    }

 消息处理线程定义代码：

import com.bohuikeji.frame.module.admin.base.MqListener;

/**
 * 消息队列消费者业务处理线程
 * @author sjh
 */
public class MqListenerThread implements Runnable{

    private MqListener listener;
    private Object msg;

    public MqListenerThread(MqListener listener, Object msg){
        this.listener=listener;
        this.msg=msg;
    }

    @Override
    public void run() {
        try {
            listener.onReceive(msg);
        }catch (Exception e){
            log.error("【消息队列监听者业务异常】位于：{},错误信息：{}",listener.getClass(),e.getMessage(),e);
            listener.onError(msg,e);
        }
    }
}

SpringUtil工具类的代码：

import org.springframework.beans.BeansException;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.stereotype.Component;

/**
 * SpringBean工具类
 * @author sjh
 */
@Component
public class SpringUtil implements ApplicationContextAware {
    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        if (SpringUtil.applicationContext == null) {
            SpringUtil.applicationContext = applicationContext;
        }
    }

    /**
     * applicationContext
     * @return
     */
    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    /**
     * 通过name获取 Bean.
     * @param name beanName
     * @return
     */
    public static Object getBean(String name) {
        return getApplicationContext().getBean(name);
    }

    /**
     * 通过class获取Bean.
     * @param clazz beanClass
     * @param <T>
     * @return
     */
    public static <T> T getBean(Class<T> clazz) {
        return getApplicationContext().getBean(clazz);
    }

    /**
     * 通过name,以及Clazz返回指定的Bean
     * 【shioy】
     * @param name beanName
     * @param clazz beanClass
     * @param <T>
     * @return
     */
    public static <T> T getBean(String name, Class<T> clazz) {
        return getApplicationContext().getBean(name, clazz);
    }

    /**
     * 手动注册Bean
     * @param requiredType 待注册的BeanClass文件
     * @param <T> bean类型
     * @return 返回Bean对象
     */
    public static  <T> T registerBean(Class<T> requiredType){
        return registerBean(requiredType,requiredType.getSimpleName());
    }

    /**
     *  动态注入bean
     * @param requiredType 注入类
     * @param beanName bean名称
     */
    public static <T> T registerBean(Class<T> requiredType,String beanName){
        //将applicationContext转换为ConfigurableApplicationContext
        ConfigurableApplicationContext configurableApplicationContext = (ConfigurableApplicationContext) applicationContext;
        //获取BeanFactory
        DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) configurableApplicationContext.getAutowireCapableBeanFactory();
        //创建bean信息.
        BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(requiredType);
        //动态注册bean.
        defaultListableBeanFactory.registerBeanDefinition(beanName, beanDefinitionBuilder.getBeanDefinition());
        //获取动态注册的bean.
        return configurableApplicationContext.getBean(requiredType);
    }

    /**
     * 读取当前配置环境
     * @return
     */
    public static String getActiveProfile() {
        return applicationContext.getEnvironment().getActiveProfiles()[0];
    }


    /**
     * 读取环境配置文件
     * @param propertyName
     * @return
     */
    public static String getProperty(String propertyName) {
        return applicationContext.getEnvironment().getProperty(propertyName);
    }

}

三、使用

使用仅需2步

1、定义编写消费者代码，也就是继承MqListener<T>类，并实现。

2、在需要发消息的调用Mqtool类的方法，发送消息。

示例如下：

这是消费者业务定义代码(可以理解为消息监听者)：

import com.bohuikeji.frame.module.admin.base.MqListener;
import org.springframework.stereotype.Service;

/**
 * RabbitMq封装使用：消费者业务代码示例
 * @author Admin
 */
@Service
public class TestListener extends MqListener<String> {
    @Override
    public boolean onReceive(String msg) {
        System.out.println("TestListener接收到消息的："+msg);
        return true;
    }

    @Override
    public void onError(String msg, Exception e) {
        System.out.println("业务代码出现异常，可在此处进行处理！");

    }
}

这是发送消息的代码，示例发送了三种类型消息：

@Autowired MqTool mqTool;

    /**
     * 消息发送测试
     * @return
     */
    @GetMapping("/test")
    public void test(){
        //定义字符串消息
        String msg="这是一条字符串消息";

        //发送普通消息：集群环境下，仅有一个消费者会收到消息
        mqTool.sendMsg(msg,TestListener.class);

        //发送广播消息：集权环境下，所有的集群节点(消费者)都会收到消息
        mqTool.sendMsgToAll(msg, TestListener.class);

        //发送延迟消息：延时20秒后，仅有一个消费者会收到消息
        mqTool.sendDelayMsg(msg,TestListener.class, Duration.ofSeconds(20));
    }

补充：如需发送其他类型消息，如Ineger，实体包装类等，则定义消费者代码时候指定泛型<T>的类型即可，示例如下：

@Service
public class TestListener extends MqListener<SysUser> {
    @Override
    public boolean onReceive(SysUser user) {
        System.out.println("TestListener接收到消息的："+user);
        return true;
    }

    @Override
    public void onError(SysUser user, Exception e) {
        System.out.println("业务代码出现异常，可在此处进行处理！");

    }
}

四、结束

封装后的代码已经在笔者的项目中投入生产使用了，后续可能会再消息被动重试、主动重试、以及异常处理方面再做加强，如果您有好的意见或想法，欢迎留言评论。