526互联

SpringBoot集成多个RabbitMq（多个MQ链接）

发布时间 2023-12-16 23:22:22作者: 蓝色土耳其

##2023年12月16日 20:25:36  项目中使用RabbitMQ作为应用间信息互通，本次梳理下关于MQ的使用。

1、引入依赖

<!--  引入依赖，使用v2.5.6版本 -->
<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.5.6</version>
        </dependency>
</dependencies>

2、在Nacos中配置MQ信息，此次项目中使用了两个MQ实例（①已方提供MQ Server，本项目内各系统信息互通；②甲方提供MQ Server，与甲方做信息互通），本次介绍mq1

spring:
  rabbitmq:
    mq1:
      host: 127.0.0.1
      port: 5672
      username: admin
      password: ****
      enable: false ##队列是否启用，在Configuration中来确认是否初始化MQ
    mq2:
      host: 127.0.0.2
      port: 5672
      username: admin
      password: ***
      enable: false

3、编写配置文件

@Data
@Component("mq1RabbitmqConfig") 
@ConfigurationProperties(prefix = "spring.rabbitmq.mq1") //读取mq1的配置信息
@ConditionalOnProperty(name = "spring.rabbitmq.mq1.enable", havingValue = "true") //是否启用
public class MQ1RabbitConfiguration {
    private String host;
    private Integer port;
    private String username;
    private String password;

    @Autowired
    private ReturnCallBack returnCallBack;
    @Autowired
    private ConfirmCallBack confirmCallBack;

    @Bean(value = "mq1ConnectionFactory")   //命名mq1的ConnectionFactory,如果项目中只有一个mq则不必如此
    public ConnectionFactory createConnectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        //开启发送到交换机和队列的回调
        connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        return connectionFactory;
    }
    @Bean(name = "mq1RabbitTemplate") //命名mq1的RabbitTemplate,如果项目中只有一个mq则不必如此
    public RabbitTemplate brainRabbitTemplate(@Qualifier("mq1ConnectionFactory") ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //发送消息时设置强制标志，仅当提供了returnCallback时才适用
        rabbitTemplate.setMandatory(true);
        //确保消息是否发送到交换机，成功与失败都会触发
        rabbitTemplate.setConfirmCallback(confirmCallBack);
        //确保消息是否发送到队列，成功发送不触发，失败触发
        rabbitTemplate.setReturnsCallback(returnCallBack);
        return rabbitTemplate;
    }
    @Bean("mq1RabbitListenerContainerFactory")//命名mq1的RabbitListenerContainerFactory,如果项目中只有一个mq则不必如此
    SimpleRabbitListenerContainerFactory mq1RabbitListenerContainerFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("mq1ConnectionFactory") ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory listenerContainerFactory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(listenerContainerFactory, connectionFactory);
        return listenerContainerFactory;
    }
}

4、编写回调callback,此一举措是为了记录交换机、队列本身是否健康，如业务无此细纠，也可不记录。

ConfirmCallBack.java ：消息发送到交换机成功、失败都会回调
ReturnCallBack.java：消息发送队列失败回调
@Component
@Slf4j
public class ConfirmCallBack implements RabbitTemplate.ConfirmCallback {
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if(!ack){
            log.info("消息发送交换机失败：{}",s);
        }else{
            log.info("消息发送交换机成功");
        }
    }
}

@Component
@Slf4j
public class ReturnCallBack implements RabbitTemplate.ReturnsCallback {
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        log.info("消息发送队列失败：｛｝", JSON.toJSON(returnedMessage));
    }
}

 

5、MQ工具类，方便各业务点调用，减少代码冗余。

@Component
@ConditionalOnProperty(name = "spring.rabbitmq.mq1.enable", havingValue = "true") //是否启用
public class RabbitMQUtil {

    @Resource(name = "mq1RabbitTemplate")  //初始化mq1的RabbitTemplate对象，如果项目中只有一个MQ，则无需这么麻烦，
    private RabbitTemplate mq1RabbitTemplate;

    public void test1(List ob){
        if(ObjectUtil.isNotEmpty(pos)){
          //TODO
        }
    }
    public void test2(String str){
          //TODO
    }
}

　　

6、生产者Producer

@Service
public class producer {

    @Autowired(required = false)  //一定要加required=false，否则mq配置不启用时，这里会报错
    private RabbitMQUtil rabbitMQUtil;

    public void producer(List ob) {
        
        if(rabbitMQUtil != null){
            rabbitMQUtil.test1(ob);
        }
    }
}

7、消费者Consumer（Listener）

①此处采用动态绑定生成队列的方式，可以满足多据点的模式，部署A据点后，根据A的特征值自动创建关于A的exchage、queue(mq1Queue_A)，免去了在java类中显式声明并绑定exchage和queue的代码

②此处使用了死信队列 x-dead-letter-exchange，可以在msg消费失败时让msg进入对应死信队列，进而监听死信队列进行补偿操作。

@Slf4j
@Component
@ConditionalOnExpression(value = "${spring.rabbitmq.mq1.enable:false} && '${aa.bb}' eq 'cc' ")  //mq1队列启用，并且满足某些条件才初始化
public class MQ1ccRabbitListener {
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "mq1Queue_A"
                    , arguments = {@Argument(name = "x-dead-letter-exchange", value = "dead.letter.exchange"),
                    @Argument(name = "x-dead-letter-routing-key", value = "dead.letter.routekey")}
　　　　　　　　),
            exchange = @Exchange(name = "mq1_fanout_exchange",type = ExchangeTypes.FANOUT)
    ),ackMode = "MANUAL")
    public void mq1listener1(Message message,Channel channel, String poJson) throws IOException {
        try {
            MyClass po = JSONUtil.toBean(poJson,MyClass.class);
            if(myService.saveOrUpdate(po)){
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }else {
                //第二个参数为false时，进入死信队列消费，true时重新进入队列头
                channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
            }
        } catch (Exception e) {
            log.info("队列{}消费省级目录信息失败：{},异常信息：{}",RabbitMqConstants.DOWN_REGION_QUEUES + "_${unified.areaCode}",poJson,e.getMessage());
            //第二个参数为false时，进入死信队列消费，true时重新进入队列头
            channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
        }
    }
}

 

8、显式绑定死信队列的exchange和queue（其他队列也可以这样绑，死信队列和其他队列没有任何区别，只是说可以在普通队列中去用 x-dead-letter-exchange 去标记一个队列的死信队列，让普通队列消费失败时，将消息扔到死信队列。

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.mq1.enable", havingValue = "true")
public class BrainDiedConfig {

    @Resource(name = "mq1RabbitTemplate")
    private RabbitTemplate brainRabbitTemplate;
    @Bean("dead.letter.exchange")
    public Exchange diedExchange(){
        return new DirectExchange("dead.letter.exchange");
    }

    @Bean("dead.letter.queues")
    public Queue diedQueue(){
        return new Queue("dead.letter.queues",true);
    }

    @Bean
    public Binding bindingDied(@Qualifier("dead.letter.queues") Queue diedQueue,@Qualifier("dead.letter.exchange") Exchange diedExchange){
        return BindingBuilder.bind(diedQueue).to(diedExchange).with("dead.letter.routeke").noargs();
    }
}

9、消费死信队列

    @RabbitListener(queues = "dead.letter.queues")
    public void diedLetterConsumer(Message message, Channel channel, String poJson){
        try {
            String queueName = (String) message.getMessageProperties().getHeaders().get("x-first-death-queue");
            String exchangeName = (String) message.getMessageProperties().getHeaders().get("x-first-death-exchange");
            dao.save(queueName, exchangeName, poJson);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

10、死信队列建表语句

create table died_letter
(
    id            bigint auto_increment
        primary key,
    queue_name    varchar(50)                        not null comment '队列名称'
    exchange_name varchar(50)                        not null comment '交换机名称',
    msg_json      mediumtext                         not null comment '队列信息',
    create_time   datetime default CURRENT_TIMESTAMP not null comment '生成时间',
    is_handle     tinyint  default 0                 not null comment '是否被处理，0：未处理；1：已处理'
)
    comment '死信队列信息';