SpringBoot配置多个RabbitMq「建议收藏」

大家好，欢迎来到IT知识分享网。

• YML

 rabbitmq:
    first:
      username: ${app.appkey}
      password: ${app.appkey}
      virtual-host: ${app.appid}
      addresses: x.x.x.x:5672,x.x.x.x:5672 #集群
    second:
      username: guest 
      password: guest
      virtual-host: /
      host: 127.0.0.1
      port: 5672

IT知识分享网

• 配置源

IT知识分享网import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

/**
 * RabbitMq多源配置
 *
 * @author Lenovo
 */
@Configuration
public class RabbitConfig {

    @Bean(name = "firstConnectionFactory")
    @Primary
    public ConnectionFactory firstConnectionFactory(
            @Value("${spring.rabbitmq.first.addresses}") String addresses,
            @Value("${spring.rabbitmq.first.username}") String username,
            @Value("${spring.rabbitmq.first.password}") String password,
            @Value("${spring.rabbitmq.first.virtual-host}") String virtualHost
    ) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();

        connectionFactory.setAddresses(addresses);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }

    @Bean(name = "secondConnectionFactory")
    public ConnectionFactory secondConnectionFactory(
            @Value("${spring.rabbitmq.second.host}") String host,
            @Value("${spring.rabbitmq.second.port}") int port,
            @Value("${spring.rabbitmq.second.username}") String username,
            @Value("${spring.rabbitmq.second.password}") String password,
            @Value("${spring.rabbitmq.second.virtual-host}") String virtualHost
    ) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setHost(host);
        connectionFactory.setPort(port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        return connectionFactory;
    }

    @Bean(name = "firstRabbitTemplate")
    @Primary
    public RabbitTemplate firstRabbitTemplate(
            @Qualifier("firstConnectionFactory") ConnectionFactory connectionFactory
    ) {
        RabbitTemplate firstRabbitTemplate = new RabbitTemplate(connectionFactory);
        return firstRabbitTemplate;
    }

    @Bean(name = "secondRabbitTemplate")
    public RabbitTemplate secondRabbitTemplate(
            @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory
    ) {
        RabbitTemplate secondRabbitTemplate = new RabbitTemplate(connectionFactory);
        return secondRabbitTemplate;
    }


    @Bean(name = "firstFactory")
    public SimpleRabbitListenerContainerFactory firstFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("firstConnectionFactory") ConnectionFactory connectionFactory
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }

    @Bean(name = "secondFactory")
    public SimpleRabbitListenerContainerFactory secondFactory(
            SimpleRabbitListenerContainerFactoryConfigurer configurer,
            @Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory
    ) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        return factory;
    }
}

• 信道构建器

import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.io.IOException;

/**
 * 信道构建器
 *
 * @author Lenovo
 */
@Configuration
public class CreateQueue {

    @Bean
    public String chargeQueue(@Qualifier("secondConnectionFactory") ConnectionFactory connectionFactory) {
        try {
            connectionFactory.createConnection().createChannel(false).queueDeclare("test.add", true, false, false, null);
        }catch (IOException e){
            e.printStackTrace();
        }
        return "test.add";
    }
}

• 信道监听器

IT知识分享网package com.ciih.authcenter.client.mq;

import com.ciih.authcenter.manager.entity.Permission;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;

/**
 * 信道监听器
 *
 * @author Lenovo
 */
@Slf4j
@Component
public class ListeningHandle {

    public static final String ENCODING = "UTF-8";

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.USERS_ADD}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageUserAdd(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息userAdd] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.USERS_UPDATE}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageUserUpdate(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息userUpdate] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.USERS_DELETE}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageUserDelete(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息userDelete] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.ORGS_ADD}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageOrgsAdd(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息orgsAdd] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.ORGS_UPDATE}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageOrgsUpdate(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息orgsUpdate] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitHandler
    @RabbitListener(queues = {RabbitConfig.ORGS_DELETE}, containerFactory = "firstFactory")
    @SneakyThrows
    public void onMessageOrgsDelete(Message message, Channel channel) {
        log.info("[listenerManualAck 监听的消息orgsDelete] - [{}]", new String(message.getBody(), ENCODING));
        try {
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (
                IOException e) {
        }
    }

    @RabbitListener(queues = {"test.add"}, containerFactory = "secondFactory")
    @SneakyThrows
    public void hospitalAdd(List<Permission> permissions, Message message, Channel channel) {
        System.out.println(permissions);
    }
}

• 发送消息

import com.ciih.authcenter.manager.entity.Permission;
import com.ciih.authcenter.manager.service.PermissionService;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.List;

@RestController
public class Sender {

    @Resource
    PermissionService permissionService;

    @Resource(name = "secondRabbitTemplate")
    private RabbitTemplate secondRabbitTemplate;

    @GetMapping("test1")
    public void send1() {
        List<Permission> list = permissionService.lambdaQuery().last("limit 0, 10").list();
        this.secondRabbitTemplate.convertAndSend("test.add", list);
    }
}

• 依赖

  <dependency>
              <groupId>org.springframework.boot</groupId>
              <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>