自定义springboot组件--基于模板模式对原生springboot的rabbitmq组件进行扩展

在我们的日常开发中,消息中间件已经成为了java研发工程师的一项必备技能,本文主要是基于对springboot原生组件的扩展开发,基于模板设计模式和静态代理模式,简化了队列路由的绑定,交由公共模板进行统一的绑定,并在公用模板中保证了消息的幂等性和消息的可靠性投递,将这些类似的代码抽离出来,让开发者只专注于业务逻辑的开发.

整体实现思路:

• 开发者申明路由交换机等基础元数据后,交由元数据解析器完成交换机,路由的申明及相应关系的绑定
• 开发者申明消息监听器后,通过消息侦听容器完成队列与监听器的绑定
• 开发者通过消息发射器,发布消息后,系统通过不通模板完成消息的发送

申明基础信息接口

public interface MessageMetaData {
    
    String getQueue();

    
    ExchangeTypeEnum getExchangeType();

    
    default Map getQueueArgs(){
        return null;
    };

    
    default Map getExchangeArgs(){
        return null;
    };

    
    default MessageProperties getMessageProperties(){
        return null;
    };

    default void setMessageProperties(MessageProperties messageProperties){

    };

}

定义不同的元数据模板

public abstract class FanoutMessageMetaData implements MessageMetaData {

    private MessageProperties messageProperties = null;

    @Override
    public ExchangeTypeEnum getExchangeType() {
        return ExchangeTypeEnum.FANOUT;
    }

    
    abstract public String getExchange();


    @Override
    public MessageProperties getMessageProperties() {
        return this.messageProperties;
    }

    @Override
    public void setMessageProperties(MessageProperties messageProperties) {
        this.messageProperties=messageProperties;
    }
}

public abstract class DirectMessageMetadata implements MessageMetaData{
    private MessageProperties messageProperties;
    @Override
    public ExchangeTypeEnum getExchangeType() {
        return ExchangeTypeEnum.DIRECT;
    }

    
    abstract public String getExchange();

    @Override
    public MessageProperties getMessageProperties() {
        return this.messageProperties;
    }

    @Override
    public void setMessageProperties(MessageProperties messageProperties) {
        this.messageProperties=messageProperties;
    }
}

申明元数据解析接口,定义解析规范

public interface Support {
    Boolean support(ExchangeTypeEnum exchangeTypeEnum);
}


public interface MessageMetaDataResolver extends Support {
    
    void resolve(MessageMetaData messageMetaData);
}

定义不同模板的解析器

@RequiredArgsConstructor
@Slf4j
public abstract class AbstractMessageMetadataResolver implements MessageMetaDataResolver {
    private final RabbitAdmin rabbitAdmin;


    @Override
    public void resolve(MessageMetaData messageMetaData) {
        if (support(messageMetaData.getExchangeType())){
            doResolve(messageMetaData);
        }
    }

    
    abstract void doResolve(MessageMetaData messageMetaData);

    
    public void declareQueue(Queue queue){
        rabbitAdmin.declareQueue(queue);
        log.info("queue [{}] declared.",queue);
    }

    public void declareExchange(Exchange exchange){
        rabbitAdmin.declareExchange(exchange);
        log.info("exchange [{}] declared.",exchange);
    }

    public void declareBinding(Binding binding){
        rabbitAdmin.declareBinding(binding);
        log.info("binding [{}] declared.",binding);
    }

}
public class DirectMessageMetadataResolver extends AbstractMessageMetadataResolver{

    public DirectMessageMetadataResolver(RabbitAdmin rabbitAdmin) {
        super(rabbitAdmin);
    }

    @Override
    public Boolean support(ExchangeTypeEnum exchangeTypeEnum) {
        return ExchangeTypeEnum.DIRECT.equals(exchangeTypeEnum);
    }

    @Override
    void doResolve(MessageMetaData messageMetaData) {
        DirectMessageMetadata directMessageMetadata = (DirectMessageMetadata) messageMetaData;

        // 申明队列
        Queue queue = new Queue(directMessageMetadata.getQueue(), true, false, false, directMessageMetadata.getQueueArgs());
        declareQueue(queue);

        // 申明交换机
        DirectExchange exchange = new DirectExchange(directMessageMetadata.getExchange(), true, false, directMessageMetadata.getExchangeArgs());
        declareExchange(exchange);

        // 交换机绑定队列
        Binding binding = BindingBuilder.bind(queue).to(exchange).withQueueName();
        declareBinding(binding);
    }
}

public class FanoutMessageMetaDateResolver extends AbstractMessageMetadataResolver{
    public FanoutMessageMetaDateResolver(RabbitAdmin rabbitAdmin) {
        super(rabbitAdmin);
    }

    @Override
    public Boolean support(ExchangeTypeEnum exchangeTypeEnum) {
        return ExchangeTypeEnum.FANOUT.equals(exchangeTypeEnum);
    }

    @Override
    void doResolve(MessageMetaData messageMetaData) {
        FanoutMessageMetaData FanoutMessageMetaData = (FanoutMessageMetaData) messageMetaData;
        // 申明队列
        Queue queue = new Queue(FanoutMessageMetaData.getQueue(), true, false, false, FanoutMessageMetaData.getQueueArgs());
        declareQueue(queue);

        // 申明交换机
        FanoutExchange exchange = new FanoutExchange(FanoutMessageMetaData.getExchange(), true, false, FanoutMessageMetaData.getExchangeArgs());
        declareExchange(exchange);

        // 队列绑定交换机
        Binding binding = BindingBuilder.bind(queue).to(exchange);
        declareBinding(binding);

    }
}

申明静态代理器

@Slf4j
public class DelegatingMessageMetaDataResolver implements MessageMetaDataResolver {

    private final Map messageMetaDataResolverMap = new ConcurrentHashMap<>();

    public DelegatingMessageMetaDataResolver(Map messageMetaDataResolverMap){
       this.messageMetaDataResolverMap.putAll(messageMetaDataResolverMap);
    }

    @Override
    public Boolean support(ExchangeTypeEnum exchangeTypeEnum) {
        return true;
    }

    @Override
    public void resolve(MessageMetaData messageMetaData) {
        if (messageMetaData==null){
            log.error("metaDate resolve must have messageMetaData but is null");
            return;
        }
        MessageMetaDataResolver messageMetaDataResolver = messageMetaDataResolverMap.get(messageMetaData.getExchangeType());
        if (messageMetaDataResolver==null){
            log.error("messageMetaDataResolver is null");
            return;
        }
        messageMetaDataResolver.resolve(messageMetaData);
    }
}

元数据初始化器

@RequiredArgsConstructor
@Slf4j
public class MessageMetaDataInitalizer implements InitializingBean, Ordered {

    private final XlcpMqProperties xlcpMqProperties;

    private final MessageMetaDataResolver messageMetaDataResolver;

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("autoCreateMq configured with [{}]",xlcpMqProperties.getAutoCreateMq());
        if (xlcpMqProperties.getAutoCreateMq()){
            Map messageMetaDataMap = SpringContextHolder.getBeansOfType(MessageMetaData.class);
            messageMetaDataMap.forEach((key,messageMetaData)->{
                log.info("Mq auto declared with metaDate [{}]",messageMetaData);
                messageMetaDataResolver.resolve(messageMetaData);
            });
        }
    }


    @Override
    public int getOrder() {
        return Ordered.LOWEST_PRECEDENCE;
    }
}

将静态代理器和初始器交由spring容器管理

申明抽象消息监听器

@Slf4j
public abstract class DynamicMessageListener extends AbstractAdaptableMessageListener {
    private final MessageMetaData messageMetaData;

    public static final String MQ_MESSAGE_ID_PREFIX = "mq:message:id";

    public DynamicMessageListener(MessageMetaData messageMetaData){
        this.messageMetaData=messageMetaData;
    }

    public void doExcute(Message message, Channel channel, Consumer