手写超级好用的rabbitmq-spring-boot-start启动器

这篇具有很好参考价值的文章主要介绍了手写超级好用的rabbitmq-spring-boot-start启动器。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

手写超级好用的rabbitmq-spring-boot-start启动器

1.前言

  由于springBoot官方提供的默认的rabbitMq自动装配不是那么好用,一个项目中只能配置使用一个rabbitMq的服务器,队列也需要编码的方式定义,这种繁杂且不易使用,用一次需要写一次硬编码,之前有一个想法是,能不能使用springBoot官方提供的自动装配实现一个多rabbitMq多队列配置并且支持多种延迟队列的这种多对多关系的实现,但是左思右想,springBoot官方提供的这个rabbitMq自动装配不能满足我的需求,所以我在酝酿了很久,也把官方那个自动装配的源码看了一遍又一遍,随着我之前手写实现了好几个starter启动器,然后就想实现一个rabbitMq的starter,只要简单的配置即可轻松的实现上面的功能,然后提供了一套好用的api,使用的时候只需要在项目中引入这个启动器,节省很大的硬编码和配置灵活,配置改变只需要重启项目即可,对业务使用友好的一个starter启动器,再也不用为如何使用rabbitMq的集成而烦恼了,只需要简单的配置就可以实现好用的功能,让我们把精力放在业务上,而不是代码和代码集成上,大大的提升开发效率和节省我们宝贵的时间,让我们用宝贵的时间来享受时光,生命和生活,效率至上,远离加班,简约也简单,优雅也优美,简单配置就可以实现交换机、队列、绑定关系等根据配置自动装配,然后实现发送普通消息和3种延迟队列发送延迟消息,3中延迟队列实现如下:

  一:延迟插件实现延迟队列

    交换机类型必须CustomExchange

  二:TTL + 死信队列/延迟交换机实现延迟队列

  三: 延迟交换机 + 消息设置setHeader(“x-delay”, xxx)

  还可以配置相同的rabbitMq服务器不同的虚拟机,单独配置,遵循下标递增不重复即可

实现思路如下。

2.工程目录结构

手写超级好用的rabbitmq-spring-boot-start启动器,rabbitmq,RabbitMq启动器

3.主要实现原理

3.1spring.factories配置

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  com.zlf.config.RabbitConfig,\
  com.zlf.config.ExchangeQueueConfig,\
  com.zlf.starter.RabbitAutoConfiguration

3.2EnableZlfRabbitMq配置

package com.zlf.starter;

import org.springframework.context.annotation.Import;

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * 使用需要在启动类上加入@EnableZlfRabbit注解
 * 启动类上排除默认的自动装配RabbitAutoConfiguration
 *
 * @author zlf
 * 启动类上加入如下配置
 * @SpringBootApplication(exclude = {
 * RabbitAutoConfiguration.class})
 * @Import(value = {RabbitService.class, ZlfMqSpringUtils.class})
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Import(ZlfRabbitMqRegistrar.class)
public @interface EnableZlfRabbitMq {

}

3.3RabbitAutoConfiguration配置

package com.zlf.starter;

import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.config.RabbitListenerConfigUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Configuration;

@Configuration(proxyBeanMethods = false)
@EnableRabbit
@ConditionalOnMissingBean(name = RabbitListenerConfigUtils.RABBIT_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)
public class RabbitAutoConfiguration {

}

3.4ZlfRabbitMqRegistrar配置

package com.zlf.starter;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.zlf.config.ExchangeQueueConfig;
import com.zlf.config.ExchangeQueueProperties;
import com.zlf.config.RabbitConfig;
import com.zlf.config.RabbitProperties;
import com.zlf.config.RabbitProperties.AmqpContainer;
import com.zlf.config.RabbitProperties.Cache;
import com.zlf.config.RabbitProperties.Cache.Connection;
import com.zlf.config.RabbitProperties.ContainerType;
import com.zlf.config.RabbitProperties.DirectContainer;
import com.zlf.config.RabbitProperties.ListenerRetry;
import com.zlf.config.RabbitProperties.Retry;
import com.zlf.config.RabbitProperties.SimpleContainer;
import com.zlf.config.RabbitProperties.Template;
import com.zlf.constants.ErrorExchangeQueueInfo;
import com.zlf.dto.ExchangeQueueDto;
import com.zlf.enums.DelayTypeEnum;
import com.zlf.enums.ExchangeTypeEnum;
import com.zlf.enums.FunctionTypeEnum;
import com.zlf.service.RabbitService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.AbstractExchange;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.config.DirectRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.connection.RabbitConnectionFactoryBean;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.amqp.rabbit.support.ValueExpression;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.BeansException;
import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.beans.factory.config.ConstructorArgumentValues;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.core.type.AnnotationMetadata;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * spring:
 * rabbitmq:
 * listener:
 * simple:
 * acknowledge-mode: auto #由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack
 * manual:手动ack,需要在业务代码结束后,调用api发送ack。
 * auto:自动ack,由spring监测listener代码是否出现异常,没有异常则返回ack;抛出异常则返回nack
 * none:关闭ack,MQ假定消费者获取消息后会成功处理,因此消息投递后立即被删除(此时,消息投递是不可靠的,可能丢失)
 * <p>
 * 原文链接:https://blog.csdn.net/m0_53142956/article/details/127792054
 *
 * @author zlf
 */
@Slf4j
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass({RabbitTemplate.class, Channel.class})
@EnableConfigurationProperties(RabbitConfig.class)
public class ZlfRabbitMqRegistrar implements ImportBeanDefinitionRegistrar, EnvironmentAware, BeanFactoryAware {

    private BeanFactory beanFactory;
    private RabbitConfig rabbitConfig;
    private ExchangeQueueConfig exchangeQueueConfig;

    @SneakyThrows
    @Override
    public void registerBeanDefinitions(AnnotationMetadata annotationMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {
        List<RabbitProperties> rps = rabbitConfig.getRps();
        if (CollectionUtils.isEmpty(rps)) {
            throw new RuntimeException("rabbitMq的rps配置不为空,请检查配置!");
        }
        log.info("zlf.registerBeanDefinitions:rps.size:{},rps:{}", rps.size(), JSON.toJSONString(rps));
        List<ExchangeQueueProperties> eqps = exchangeQueueConfig.getEqps();
        if (CollectionUtils.isEmpty(eqps)) {
            throw new RuntimeException("rabbitMq的eqps配置不为空,请检查配置!");
        }
        log.info("zlf.registerBeanDefinitions:eqps.size:{},rps:{}", eqps.size(), JSON.toJSONString(eqps));
        for (int i = 0; i < rps.size(); i++) {
            this.checkRabbitProperties(rps.get(i));
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(getRabbitConnectionFactoryBean(rps.get(i)).getObject());
            cachingConnectionFactory.setAddresses(rps.get(i).determineAddresses());
            cachingConnectionFactory.setPublisherReturns(rps.get(i).getPublisherReturns());
            cachingConnectionFactory.setPublisherConfirmType(rps.get(i).getPublisherConfirmType());
            Cache.Channel channel = rps.get(i).getCache().getChannel();
            if (Objects.nonNull(channel.getSize())) {
                cachingConnectionFactory.setChannelCacheSize(channel.getSize());
            }
            if (Objects.nonNull(channel.getCheckoutTimeout())) {
                Duration checkoutTimeout = channel.getCheckoutTimeout();
                cachingConnectionFactory.setChannelCheckoutTimeout(checkoutTimeout.toMillis());
            }
            Connection connection = rps.get(i).getCache().getConnection();
            if (Objects.nonNull(connection.getMode())) {
                cachingConnectionFactory.setCacheMode(connection.getMode());
            }
            if (Objects.nonNull(connection.getSize())) {
                cachingConnectionFactory.setConnectionCacheSize(connection.getSize());
            }
            // 注册cachingConnectionFactory的bean
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(CachingConnectionFactory.class.getName() + i, cachingConnectionFactory);
            log.info("zlf.ConfigurableBeanFactory注册完成,beanName:{}", CachingConnectionFactory.class.getName() + i);
            // 注册RabbitAdmin的bean
            RabbitAdmin rabbitAdmin = new RabbitAdmin(cachingConnectionFactory);
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(RabbitAdmin.class.getName() + i, rabbitAdmin);
            log.info("zlf.RabbitAdmin注册完成,beanName:{}", RabbitAdmin.class.getName() + i);
            //构建发送的RabbitTemplate实例关联连接工厂
            Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
            RabbitTemplate rabbitTemplate = new RabbitTemplate(cachingConnectionFactory);
            rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
            Template template = rps.get(i).getTemplate();
            ConfirmType publisherConfirmType = rps.get(i).getPublisherConfirmType();
            log.info("第{}个配置的publisherConfirmType:{}", i, JSON.toJSONString(publisherConfirmType));
            //生产者confirm
            /**
             * publish-confirm-type:开启publisher-confirm,这里支持两种类型:
             * simple:【同步】等待confirm结果,直到超时(可能引起代码阻塞)
             * correlated:【异步】回调,定义ConfirmCallback,MQ返回结果时会回调这个ConfirmCallback
             * publish-returns:开启publish-return功能,同样是基于callback机制,不过是定义ReturnCallback
             * template.mandatory:
             * 定义当消息从交换机路由到队列失败时的策略。
             * 【true,则调用ReturnCallback;false:则直接丢弃消息】
             */
            if (ConfirmType.CORRELATED.equals(publisherConfirmType)) {
                rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
                    if (Objects.nonNull(correlationData)) {
                        if (Objects.nonNull(ack) && ack) {
                            log.info("消息发送成功->correlationData:{}", JSON.toJSONString(correlationData));
                        } else if (StringUtils.isNotBlank(cause)) {
                            log.error("消息->correlationData:{}->发送失败原因->{}", JSON.toJSONString(correlationData), cause);
                        }
                    }
                    if (Objects.nonNull(ack) && ack) {
                        log.info("消息发送成功ack:{}", ack);
                    }
                    if (StringUtils.isNotBlank(cause)) {
                        log.error("消息发送失败原因->cause:{}", cause);
                    }
                    if (Objects.isNull(correlationData) && Objects.isNull(ack) && StringUtils.isEmpty(cause)) {
                        log.info("消息发送成功,收到correlationData,ack,cause都是null");
                    }
                });
            }
            Boolean publisherReturns = rps.get(i).getPublisherReturns();
            Boolean mandatory = template.getMandatory();
            log.info("第{}个配置的publisherReturns:{},mandatory:{}", i, publisherReturns, mandatory);
            //消息回调
            //开启强制回调
            if (mandatory && publisherReturns) {
                rabbitTemplate.setMandatory(template.getMandatory());
                rabbitTemplate.setMandatoryExpression(new ValueExpression<>(template.getMandatory()));
                //设置消息回调
                rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
                    log.error("消息->{}路由失败", message);
                    // 如果有需要的话,重发消息
                });
            }
            Retry retry = rps.get(i).getTemplate().getRetry();
            if (retry.isEnabled()) {
                RetryTemplate retryTemplate = new RetryTemplate();
                SimpleRetryPolicy policy = new SimpleRetryPolicy();
                retryTemplate.setRetryPolicy(policy);
                policy.setMaxAttempts(retry.getMaxAttempts());
                ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
                backOffPolicy.setMultiplier(retry.getMultiplier());
                if (Objects.nonNull(retry.getMaxInterval())) {
                    backOffPolicy.setMaxInterval(retry.getMaxInterval().toMillis());
                }
                rabbitTemplate.setRetryTemplate(retryTemplate);
            }
            Duration receiveTimeout = template.getReceiveTimeout();
            if (Objects.nonNull(receiveTimeout)) {
                rabbitTemplate.setReceiveTimeout(receiveTimeout.toMillis());
            }
            Duration replyTimeout = template.getReplyTimeout();
            if (Objects.nonNull(replyTimeout)) {
                rabbitTemplate.setReplyTimeout(replyTimeout.toMillis());
            }
            String exchange = template.getExchange();
            if (StringUtils.isNotBlank(exchange)) {
                rabbitTemplate.setExchange(exchange);
            }
            String routingKey = template.getRoutingKey();
            if (StringUtils.isNotBlank(routingKey)) {
                rabbitTemplate.setRoutingKey(routingKey);
            }
            String defaultReceiveQueue = template.getDefaultReceiveQueue();
            if (StringUtils.isNotBlank(defaultReceiveQueue)) {
                rabbitTemplate.setDefaultReceiveQueue(defaultReceiveQueue);
            }
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(RabbitTemplate.class.getName() + i, rabbitTemplate);
            log.info("zlf.RabbitTemplate注册完成,beanName:{}", RabbitTemplate.class.getName() + i);
            // 不注册RabbitService
            RabbitService rabbitService = new RabbitService();
            //构建监听工厂实例并注入
            ContainerType type = rps.get(i).getListener().getType();
            if (ContainerType.SIMPLE.equals(type)) {
                Map<String, String> errorExchangeQueueRelationship = this.createErrorExchangeQueueRelationship(String.valueOf(i), rabbitService, rabbitAdmin);
                SimpleContainer simple = rps.get(i).getListener().getSimple();
                ConstructorArgumentValues cas3 = new ConstructorArgumentValues();
                MutablePropertyValues values3 = new MutablePropertyValues();
                this.getAmqpContainer(simple, values3, cachingConnectionFactory, jackson2JsonMessageConverter, rabbitTemplate, errorExchangeQueueRelationship);
                if (Objects.nonNull(simple.getConcurrency())) {
                    values3.add("concurrentConsumers", simple.getConcurrency());
                }
                if (Objects.nonNull(simple.getMaxConcurrency())) {
                    values3.add("maxConcurrentConsumers", simple.getMaxConcurrency());
                }
                if (Objects.nonNull(simple.getBatchSize())) {
                    values3.add("batchSize", simple.getBatchSize());
                }
                RootBeanDefinition rootBeanDefinition3 = new RootBeanDefinition(SimpleRabbitListenerContainerFactory.class, cas3, values3);
                beanDefinitionRegistry.registerBeanDefinition(SimpleRabbitListenerContainerFactory.class.getName() + i, rootBeanDefinition3);
                log.info("zlf.SimpleRabbitListenerContainerFactory注册完成,beanName:{}", SimpleRabbitListenerContainerFactory.class.getName() + i);
            } else if (ContainerType.DIRECT.equals(type)) {
                Map<String, String> errorExchangeQueueRelationship = this.createErrorExchangeQueueRelationship(String.valueOf(i), rabbitService, rabbitAdmin);
                DirectContainer direct = rps.get(i).getListener().getDirect();
                ConstructorArgumentValues cas4 = new ConstructorArgumentValues();
                MutablePropertyValues values4 = new MutablePropertyValues();
                this.getAmqpContainer(direct, values4, cachingConnectionFactory, jackson2JsonMessageConverter, rabbitTemplate, errorExchangeQueueRelationship);
                if (Objects.nonNull(direct.getConsumersPerQueue())) {
                    values4.add("consumersPerQueue", direct.getConsumersPerQueue());
                }
                RootBeanDefinition rootBeanDefinition4 = new RootBeanDefinition(DirectRabbitListenerContainerFactory.class, cas4, values4);
                beanDefinitionRegistry.registerBeanDefinition(DirectRabbitListenerContainerFactory.class.getName() + i, rootBeanDefinition4);
                log.info("zlf.DirectRabbitListenerContainerFactory注册完成,beanName:{}", DirectRabbitListenerContainerFactory.class.getName() + i);
            }
            //解析注册交换机、队列和绑定关系
            ExchangeQueueProperties exchangeQueueProperties = eqps.get(i);
            log.info("zlf.exchangeQueueProperties:{}", JSON.toJSONString(exchangeQueueProperties));
            Integer index = exchangeQueueProperties.getIndex();
            log.info("zlf.exchangeQueueProperties.index:{}", index);
            if (Objects.isNull(index)) {
                throw new RuntimeException("exchangeQueueProperties.index不为空");
            }
            if (Objects.nonNull(exchangeQueueProperties)) {
                log.info("zlf.exchangeQueueProperties:{}", JSON.toJSONString(exchangeQueueProperties));
                List<ExchangeQueueDto> eqs = exchangeQueueProperties.getEqs();
                if (CollectionUtil.isNotEmpty(eqs)) {
                    for (int k = 0; k < eqs.size(); k++) {
                        String bindingIndex = index.toString() + k;
                        log.info("zlf.bindingIndex:{}", bindingIndex);
                        ExchangeQueueDto exchangeQueueDto = eqs.get(k);
                        log.info("zlf.exchangeQueueDto:{}", JSON.toJSONString(exchangeQueueDto));
                        String functionType = exchangeQueueDto.getFunctionType();
                        log.info("zlf.functionType:{}", functionType);
                        if (FunctionTypeEnum.NORMAL.getFunctionType().equals(functionType)) {
                            this.createRelationship(FunctionTypeEnum.NORMAL, exchangeQueueDto, rabbitService, rabbitAdmin, bindingIndex, false);
                        } else if (FunctionTypeEnum.DELAY.getFunctionType().equals(functionType)) {
                            Integer delayType = exchangeQueueDto.getDelayType();
                            log.info("zlf.delayType:{}", delayType);
                            if (DelayTypeEnum.ONE.getDelayType().equals(delayType)) {
                                //延迟插件实现延迟队列
                                String exchangeType = exchangeQueueDto.getExchangeType();
                                if (!ExchangeTypeEnum.CUSTOM.getExchangeType().equals(exchangeType)) {
                                    throw new RuntimeException("延迟插件实现延迟队列交换机类型exchangeType必须定义为:custom");
                                }
                                this.createRelationship(FunctionTypeEnum.DELAY, exchangeQueueDto, rabbitService, rabbitAdmin, bindingIndex, false);
                            } else if (DelayTypeEnum.TWO.getDelayType().equals(delayType)) {
                                //TTL + 死信队列实现延迟队列
                                this.createRelationship(FunctionTypeEnum.DELAY, exchangeQueueDto, rabbitService, rabbitAdmin, bindingIndex, false);
                            } else if (DelayTypeEnum.THREE.getDelayType().equals(delayType)) {
                                //延迟交换机 + 消息设置setHeader("x-delay", xxx)
                                this.createRelationship(FunctionTypeEnum.DELAY, exchangeQueueDto, rabbitService, rabbitAdmin, bindingIndex, true);
                            }
                        }
                    }
                }
            }
        }
    }

    /**
     * 检查rabbitProperties配置的主要参数
     *
     * @param rabbitProperties
     */
    private void checkRabbitProperties(RabbitProperties rabbitProperties) {
        String virtualHost = rabbitProperties.getVirtualHost();
        if (StringUtils.isEmpty(virtualHost)) {
            throw new RuntimeException("RabbitProperties.virtualHost不为空");
        }
        String addresses = rabbitProperties.getAddresses();
        if (StringUtils.isEmpty(addresses)) {
            throw new RuntimeException("RabbitProperties.addresses不为空");
        }
        Integer port = rabbitProperties.getPort();
        if (Objects.isNull(port)) {
            throw new RuntimeException("RabbitProperties.port不为空");
        }
        String username = rabbitProperties.getUsername();
        if (StringUtils.isEmpty(username)) {
            throw new RuntimeException("RabbitProperties.username不为空");
        }
        String password = rabbitProperties.getPassword();
        if (StringUtils.isEmpty(password)) {
            throw new RuntimeException("RabbitProperties.password不为空");
        }
    }

    /**
     * 创建关系
     *
     * @param functionTypeEnum
     * @param exchangeQueueDto
     * @param rabbitService
     * @param rabbitAdmin
     * @param bindingIndex
     */
    private void createRelationship(FunctionTypeEnum functionTypeEnum, ExchangeQueueDto exchangeQueueDto, RabbitService rabbitService, RabbitAdmin rabbitAdmin, String bindingIndex, Boolean isDelayed) {
        String exchangeName = exchangeQueueDto.getExchangeName();
        String exchangeType = exchangeQueueDto.getExchangeType();
        HashMap<String, Object> exchangeArgs = exchangeQueueDto.getExchangeArgs();
        log.info("zlf" + functionTypeEnum.getFunctionType() + ".exchangeName:{},exchangeType:{},exchangeArgs:{}", exchangeName, exchangeType, JSON.toJSONString(exchangeArgs));
        AbstractExchange exchange1 = rabbitService.createExchange(rabbitAdmin, exchangeName, exchangeType, exchangeArgs, isDelayed);
        exchangeName = exchangeName + bindingIndex;
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(exchangeName, exchange1);
        log.info("zlf." + functionTypeEnum.getFunctionType() + ".Exchange注册完成,beanName:{}", exchangeName);
        String queueName = exchangeQueueDto.getQueueName();
        HashMap<String, Object> queueArgs = exchangeQueueDto.getQueueArgs();
        String routingKey1 = exchangeQueueDto.getRoutingKey();
        log.info("zlf." + functionTypeEnum.getFunctionType() + ".queueName:{},queueArgs:{},routingKey1:{}", queueName, JSON.toJSONString(queueArgs), routingKey1);
        Queue queue = rabbitService.createQueue(rabbitAdmin, queueName, queueArgs);
        queueName = queueName + bindingIndex;
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(queueName, queue);
        log.info("zlf." + functionTypeEnum.getFunctionType() + ".Queue注册完成,beanName:{}", queueName);
        Binding binding = rabbitService.binding(rabbitAdmin, exchange1, queue, routingKey1);
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(Binding.class.getName() + bindingIndex, binding);
        log.info("zlf." + functionTypeEnum.getFunctionType() + ".Binding注册完成bindingIndex:{},beanName:{}", bindingIndex, Binding.class.getName() + bindingIndex);
        Integer delayType = exchangeQueueDto.getDelayType();
        if (DelayTypeEnum.TWO.getDelayType().equals(delayType)) {
            String dlxExchangeName = exchangeQueueDto.getDlxExchangeName();
            if (StringUtils.isEmpty(dlxExchangeName)) {
                throw new RuntimeException("TTL + 死信队列实现延迟队列配置参数dlxExchangeName不为空!");
            }
            String dlxExchangeType = exchangeQueueDto.getDlxExchangeType();
            if (StringUtils.isEmpty(dlxExchangeType)) {
                throw new RuntimeException("TTL + 死信队列实现延迟队列配置参数dlxExchangeType不为空!");
            }
            AbstractExchange exchange2 = rabbitService.createExchange(rabbitAdmin, dlxExchangeName, dlxExchangeType, exchangeArgs, false);
            dlxExchangeName = dlxExchangeName + bindingIndex;
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(dlxExchangeName, exchange2);
            log.info("zlf.TTL + 死信队列实现延迟队列,死信交换机注册完成,beanName:{}", dlxExchangeName);
            String dlxQueueName = exchangeQueueDto.getDlxQueueName();
            Queue queue2 = rabbitService.createQueue(rabbitAdmin, dlxQueueName, null);
            dlxQueueName = dlxQueueName + bindingIndex;
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(dlxQueueName, queue2);
            log.info("zlf.TTL + 死信队列实现延迟队列,死信队列注册完成,beanName:{}", dlxQueueName);
            String dlxKey = exchangeQueueDto.getDlxKey();
            Binding binding2 = rabbitService.binding(rabbitAdmin, exchange2, queue2, dlxKey);
            String dlxBeanName = "dlx" + Binding.class.getName() + bindingIndex + 1;
            ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(dlxBeanName, binding2);
            log.info("zlf.TTL + 死信队列实现延迟队列,死信交换机绑定队列的绑定关系注册完成,beanName:{}", dlxBeanName);
        }
    }

    private Map<String, String> createErrorExchangeQueueRelationship(String index, RabbitService rabbitService, RabbitAdmin rabbitAdmin) {
        Map<String, String> resultMap = new HashMap<>();
        String exchangeName = ErrorExchangeQueueInfo.ERROR_EXCHANGE_PREFIX + index;
        AbstractExchange exchange = rabbitService.createExchange(rabbitAdmin, exchangeName, ErrorExchangeQueueInfo.ERROR_EXCHANGE_TYPE, null, false);
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(exchangeName, exchange);
        log.info("zlf.ErrorExchange注册完成,beanName:{}", exchangeName);
        String queueName = ErrorExchangeQueueInfo.ERROR_QUEUE_PREFIX + index;
        Queue queue = rabbitService.createQueue(rabbitAdmin, queueName, null);
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(queueName, queue);
        log.info("zlf.ErrorQueue注册完成,beanName:{}", queueName);
        String errorRoutingKey = ErrorExchangeQueueInfo.ERROR_KEY_PREFIX + index;
        log.info("zlf.errorRoutingKey:{}", errorRoutingKey);
        Binding errorBinding = rabbitService.binding(rabbitAdmin, exchange, queue, errorRoutingKey);
        String errorBingBeanName = ErrorExchangeQueueInfo.ERROR_BINDING_BANE_NAME_PREFIX + Binding.class.getSimpleName() + index;
        ((ConfigurableBeanFactory) this.beanFactory).registerSingleton(errorBingBeanName, errorBinding);
        log.info("zlf.ErrorBing注册完成,beanName:{}", errorBingBeanName);
        resultMap.put("errorExchange", exchangeName);
        resultMap.put("errorRoutingKey", errorRoutingKey);
        return resultMap;
    }

    private void getAmqpContainer(AmqpContainer amqpContainer, MutablePropertyValues values, CachingConnectionFactory cachingConnectionFactory, Jackson2JsonMessageConverter jackson2JsonMessageConverter, RabbitTemplate rabbitTemplate, Map<String, String> errorExchangeQueueRelationship) {
        values.add("connectionFactory", cachingConnectionFactory);
        values.add("autoStartup", amqpContainer.isAutoStartup());
        values.add("messageConverter", jackson2JsonMessageConverter);
        if (Objects.nonNull(amqpContainer.getAcknowledgeMode())) {
            values.add("acknowledgeMode", amqpContainer.getAcknowledgeMode());
        }
        if (Objects.nonNull(amqpContainer.getPrefetch())) {
            values.add("prefetch", amqpContainer.getPrefetch());
        }
        if (Objects.nonNull(amqpContainer.getDefaultRequeueRejected())) {
            values.add("defaultRequeueRejected", amqpContainer.getDefaultRequeueRejected());
        }
        if (Objects.nonNull(amqpContainer.getIdleEventInterval())) {
            values.add("idleEventInterval", amqpContainer.getIdleEventInterval());
        }
        values.add("missingQueuesFatal", amqpContainer.isMissingQueuesFatal());
        ListenerRetry retry2 = amqpContainer.getRetry();
        if (retry2.isEnabled()) {
            RetryInterceptorBuilder<?, ?> builder = (retry2.isStateless()) ? RetryInterceptorBuilder.stateless()
                    : RetryInterceptorBuilder.stateful();
            RetryTemplate retryTemplate = new RetryTemplate();
            SimpleRetryPolicy policy = new SimpleRetryPolicy();
            retryTemplate.setRetryPolicy(policy);
            policy.setMaxAttempts(retry2.getMaxAttempts());
            ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
            backOffPolicy.setMultiplier(retry2.getMultiplier());
            if (Objects.nonNull(retry2.getMaxInterval())) {
                backOffPolicy.setMaxInterval(retry2.getMaxInterval().toMillis());
            }
            builder.retryOperations(retryTemplate);
            /**
             * 在开启重试模式后,重试次数耗尽,如果消息依然失败,则需要有MessageRecovery接口来处理,它包含三种不同的实现:
             *
             * RejectAndDontRequeueRecoverer:重试耗尽后,直接reject,【丢弃消息】【默认】就是这种方式
             * ImmediateRequeueMessageRecoverer:重试耗尽后,返回nack,消息重新入队(Immediate立刻重入队)(但是频率比没有配置消费失败重载机制低一些)
             * RepublishMessageRecoverer(推荐):重试耗尽后,将失败消息投递到指定的交换机
             */
            //消息接受拒绝后发送到异常队列
            String errorExchange = errorExchangeQueueRelationship.get("errorExchange");
            String errorRoutingKey = errorExchangeQueueRelationship.get("errorRoutingKey");
            MessageRecoverer recoverer = new RepublishMessageRecoverer(rabbitTemplate, errorExchange, errorRoutingKey);
            log.info("zlf.MessageRecoverer.errorExchange:{},errorRoutingKey:{}", errorExchange, errorRoutingKey);
            builder.recoverer(recoverer);
            values.add("adviceChain", builder.build());
        }
    }

    private RabbitConnectionFactoryBean getRabbitConnectionFactoryBean(RabbitProperties properties)
            throws Exception {
        PropertyMapper map = PropertyMapper.get();
        RabbitConnectionFactoryBean factory = new RabbitConnectionFactoryBean();
        map.from(properties::determineHost).whenNonNull().to(factory::setHost);
        map.from(properties::determinePort).to(factory::setPort);
        map.from(properties::determineUsername).whenNonNull().to(factory::setUsername);
        map.from(properties::determinePassword).whenNonNull().to(factory::setPassword);
        map.from(properties::determineVirtualHost).whenNonNull().to(factory::setVirtualHost);
        map.from(properties::getRequestedHeartbeat).whenNonNull().asInt(Duration::getSeconds)
                .to(factory::setRequestedHeartbeat);
        map.from(properties::getRequestedChannelMax).to(factory::setRequestedChannelMax);
        RabbitProperties.Ssl ssl = properties.getSsl();
        if (ssl.determineEnabled()) {
            factory.setUseSSL(true);
            map.from(ssl::getAlgorithm).whenNonNull().to(factory::setSslAlgorithm);
            map.from(ssl::getKeyStoreType).to(factory::setKeyStoreType);
            map.from(ssl::getKeyStore).to(factory::setKeyStore);
            map.from(ssl::getKeyStorePassword).to(factory::setKeyStorePassphrase);
            map.from(ssl::getTrustStoreType).to(factory::setTrustStoreType);
            map.from(ssl::getTrustStore).to(factory::setTrustStore);
            map.from(ssl::getTrustStorePassword).to(factory::setTrustStorePassphrase);
            map.from(ssl::isValidateServerCertificate)
                    .to((validate) -> factory.setSkipServerCertificateValidation(!validate));
            map.from(ssl::getVerifyHostname).to(factory::setEnableHostnameVerification);
        }
        map.from(properties::getConnectionTimeout).whenNonNull().asInt(Duration::toMillis)
                .to(factory::setConnectionTimeout);
        factory.afterPropertiesSet();
        return factory;
    }

    @Override
    public void setEnvironment(Environment environment) {
        // 通过Binder将environment中的值转成对象
        rabbitConfig = Binder.get(environment).bind(getPropertiesPrefix(RabbitConfig.class), RabbitConfig.class).get();
        exchangeQueueConfig = Binder.get(environment).bind(getPropertiesPrefix(ExchangeQueueConfig.class), ExchangeQueueConfig.class).get();
    }

    private String getPropertiesPrefix(Class<?> tClass) {
        return Objects.requireNonNull(AnnotationUtils.getAnnotation(tClass, ConfigurationProperties.class)).prefix();
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

}

4.总结

  到此,手写rabbitMq的starter实现思路就已经全部分享完了,思路比代码更重要,代码只是一个参考,用这个思路实现更多更方便简单快捷高效的轮子,制造轮子也是一种提升和给你带来成就感的事情,累并快乐着,后面我会将我之前手写的starter全部开源出来,然后将gitHub或码云地址分享给大家,制造轮子,开源给大家使用,这本身就是一种开源的精神和乐趣,Java生态最缺的就不是轮子是好用的轮子,请在看我的文章或者是转发,请把本文的原出处和作者写上去,尊重版权,创作不易,禁止原模原样的搬过去就是自己的原创,这种是不道德的行为,见到请如实举报,或者联系本作者来举报,这个starter,说实话也是构思酝酿了好久,猛干了2-3天才干出来,颈椎都干的酸,我得休息加强锻炼身体了,说实话写这个starter还是挺累的,但是搞出来的成就感满满,也方便以后集成快速使用,配置多个的rabbitMq服务器也测试过了的,也是ok的,但是配置一个rabbitMq和多个交换机、队列和绑定关系以及3种延迟队列实现是亲自测试OK的,希望我的分享能给你帮助和启发,请一键三连,么么么哒!文章来源地址https://www.toymoban.com/news/detail-850011.html

到了这里,关于手写超级好用的rabbitmq-spring-boot-start启动器的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spring Boot进阶(27):Spring Boot 整合 kafka(环境搭建+演示) | 超级详细,建议收藏

           在现代互联网应用中,消息驱动已经成为一种不可或缺的开发模式。而Kafka作为一款高性能的分布式消息系统,已经成为很多公司在消息驱动架构中的首选工具。本篇文章将介绍如何使用Spring Boot和Kafka快速构建消息驱动应用,让你在开发过程中轻松应对高并发的消息

    2024年02月05日
    浏览(45)
  • Spring Boot进阶(21):配置Undertow容器 | 超级详细,建议收藏

    环境说明:Windows10+idea2021.3.2+jdk1.8+springboot2.3.1.RELEASE         Undertow是一个高性能、灵活、轻量级的Web容器,它是Spring Boot默认的Web容器。与Tomcat相比,Undertow具有更快的启动速度、更低的内存占用,以及更好的性能和可扩展性。本篇文章将介绍如何配置Undertow容器,让你的

    2024年02月06日
    浏览(85)
  • Spring Boot进阶(52):Spring Boot 如何集成Flyway并初始化执行 SQL 脚本?| 超级详细,建议收藏

           在我们的认知中,我们会使用 SVN 或 Git 进行代码的版本管理。但是,我们是否好奇过,数据库也是需要进行版本管理的呢?         在每次发版的时候,我们可能都会对数据库的表结构进行新增和变更,又或者需要插入一些初始化的数据。而我们的环境不仅仅只

    2024年02月15日
    浏览(36)
  • 【宝藏系列】几款好用的 Spring Boot 内置工具类

    断言是一个逻辑判断,用于检查不应该发生的情况 Assert 在 JDK1.4 中引入,可通过 JVM 参数-enableassertions开启 SpringBoot 中提供了 Assert 断言工具类,通常用于数据合法性检查 获取对象的基本信息 判断工具 其他工具方法 字符串判断工具 字符串操作工具 路径相关工具方法

    2024年02月14日
    浏览(43)
  • Spring Boot入门(16):Spring Boot 整合 Swagger-UI 实现在线API接口文档 | 超级详细,建议收藏

            在现代化的软件开发中,API接口文档的编写和管理是非常重要的一环。而Swagger-UI作为一款优秀的API文档生成工具,可以帮助开发者轻松地生成并管理API接口文档,提高开发效率和代码质量。在本文中,我们将介绍如何使用Spring Boot框架和Swagger-UI工具实现在线API接

    2024年02月16日
    浏览(54)
  • [Spring boot] Spring boot 整合RabbitMQ实现通过RabbitMQ进行项目的连接

     🍳作者:天海奈奈 💭眼过千遍不如手锤一遍:推荐一款模拟面试,斩获大厂 o f f e r ,程序员的必备刷题平台 − − 牛客网  👉🏻点击开始刷题之旅 目录 什么是RabbitMQ   消息队列:接受并转发消息,类似于快递公司 消息队列的优点 消息队列的特性 RabbitMQ特点 RabbitMQ核

    2024年01月24日
    浏览(84)
  • 【Spring Boot】Spring Boot集成RabbitMQ

    Spring Boot提供了`spring-boot-starter-amqp`组件,只需要简单地配置即可与Spring Boot无缝集成。下面通过示例演示集成RabbitMQ实现消息的接收和发送。 步骤01 配置pom包。 创建Spring Boot项目并在pom.xml文件中添加spring-bootstarter-amqp等相关组件依赖: 在上面的示例中,引入Spring Boot自带的

    2024年02月06日
    浏览(50)
  • Spring Boot进阶(57):Spring中什么时候不要用@Autowired注入 | 超级详细,建议收藏

            注解@Autowired,相信对于我们Java开发者而言并不陌生吧,在SpringBoot或SpringCloud框架中使用那是非常的广泛。但是当我们使用IDEA编辑器开发代码的时候,经常会发现@Autowired 注解下面提示小黄线警告,我们把小鼠标悬停在注解上面,可以看到这个如下图所示的警告信息

    2024年02月16日
    浏览(41)
  • Spring Boot入门(04):SpringBoot实现多环境配置文件切换 | 超级详细,建议收藏

            在开发和部署Spring Boot应用的过程中,经常需要在不同的环境中进行配置,比如开发环境、测试环境、生产环境等。为了方便管理和部署,我们需要实现多环境配置文件切换。本篇教程将带你轻松搞定不同环境部署问题,让你的应用在各个环境中稳定运行。无论你是

    2024年02月12日
    浏览(54)
  • Spring Boot进阶(55):SpringBoot之集成MongoDB及实战使用 | 超级详细,建议收藏

            随着大数据时代的到来,数据存储和处理变得越来越重要。而MongoDB作为一种非关系型数据库,具有高效的数据存储和处理能力,被越来越多地应用于各种领域。尤其在Web应用开发中,SpringBoot框架已经成为了主流选择之一。在这篇文章中,我们将探讨如何将MongoD

    2024年02月17日
    浏览(48)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包