Spring AMQP + Rabbit 配置多数据源消息队列

一般在稍微大一点的项目中，需要配置多个数据库数据源，最简单的方式是用 Spring 来实现，只需要继承 AbstractRoutingDataSource 类，实现 determineCurrentLookupKey 方法，再配合使用 ThreadLocal 就可以实现。

但是如何实现 MQ 的多数据源呢？假设有部署在不同服务器上的两个消息队列，或者是同一服务器，不同 vhost 的消息队列，在一个项目中，我如何自由地选择从哪个队列收发消息呢？下面说说用 Spring AMQP + Rabbit 的实现过程及踩过的坑。

最开始的单数据源的实现很简单，网上有好多博文可以参考，官网也有介绍。主要就是创建一个 xml 的配置文件，添加各种必要的配置，声明 connection-factory、rabbitListenerContainerFactory、rabbitTemplate、queue、exchange、binding 等等。然后用 RabbitTemplate 来发消息，用 @RabbitListener 注解来监听，用 queue 指定队列来收消息，这里就不赘述了。主要说一下，在现有的基础上实现多数据源的收发。

先说配置方面，为了对比，下面先给出单数据源配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
    
    <rabbit:connection-factory id="rabbitConnectionFactory" host="${rabbit.host}" port="${rabbit.port}" username="${rabbit.username}" password="${rabbit.password}" 
　　　　　　　requested-heartbeat="30" virtual-host="${rabbit.vhost}" channel-cache-size="50"/>

    <bean id="rabbitListenerContainerFactory"
          class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="rabbitConnectionFactory"/>
        <property name="concurrentConsumers" value="16"/>
        <property name="maxConcurrentConsumers" value="50"/>
    </bean>
    
    <rabbit:admin id="rabbitAdmin" connection-factory="rabbitConnectionFactory"/>
    
    <!-- queue declare -->
    <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="queue.test"/>
    
    <!-- bind queue to exchange -->
    <rabbit:direct-exchange name="exchange" auto-delete="false" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="queue.test" key="rkey.test"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:template id="rabbitTemplate" connection-factory="rabbitConnectionFactory" retry-template="retryTemplate" reply-timeout="60000"/>

    <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
        <property name="backOffPolicy">
            <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                <property name="initialInterval" value="500"/>
                <property name="multiplier" value="10.0"/>
                <property name="maxInterval" value="10000"/>
            </bean>
        </property>
    </bean>
</beans>

为了实现双数据源，查阅了很多资料，最初实现的配置如下：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
        
    
    <rabbit:connection-factory id="rabbitConnectionFactory" host="${rabbit.host}" port="${rabbit.port}" username="${rabbit.username}"
            password="${rabbit.password}" requested-heartbeat="30" virtual-host="${rabbit.vhost}" channel-cache-size="50"/>
    
    <!-- 添加了一个连接工厂，参数从 properties 文件中取 -->
    <rabbit:connection-factory id="rabbitConnectionFactory1" host="${rabbit.host1}" port="${rabbit.port1}" username="${rabbit.username1}" 
　　　　　　　password="${rabbit.password1}" requested-heartbeat="30" virtual-host="${rabbit.vhost1}" channel-cache-size="50"/>

    
    <!-- 添加 SimpleRoutingConnectionFactory 配置，将两个 Connection factory 配置好-->
    <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory">
        <property name="targetConnectionFactories">
            <map>
                <entry key="rabbitConnectionFactory" value-ref="rabbitConnectionFactory"/>
                <entry key="rabbitConnectionFactory1" value-ref="rabbitConnectionFactory1"/>
            </map>
        </property>
    </bean>
    
    <rabbit:admin id="rabbitAdmin" connection-factory="connectionFactory"/>

    
    <!-- 由于增加了一个连接工厂，ContainerFactory 的连接工厂改为新增的 ConnectionFactory  -->
    <bean id="rabbitListenerContainerFactory"
          class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <!-- <property name="connectionFactory" ref="rabbitConnectionFactory"/> -->
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="concurrentConsumers" value="16"/>
        <property name="maxConcurrentConsumers" value="50"/>
    </bean>

    
    <!-- queue declare，增加一个消息队列 -->
    <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="queue.test"/>
    <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="queue.test1"/>
    
    <!-- bind queue to exchange -->
    <rabbit:direct-exchange name="exchange" auto-delete="false" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="queue.test" key="rkey.test"></rabbit:binding>
            <rabbit:binding queue="queue.test1" key="rkey.test1"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    
    <!-- connection-factory 改为新增的 ConnectionFactory -->
    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" retry-template="retryTemplate" reply-timeout="60000"/>

    <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
        <property name="backOffPolicy">
            <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                <property name="initialInterval" value="500"/>
                <property name="multiplier" value="10.0"/>
                <property name="maxInterval" value="10000"/>
            </bean>
        </property>
    </bean>
</beans>

改动都写在注释里了，主要就是增加了一个连接工厂的配置，其他配置做了一些相应的适配。

发消息的时候，需要指定连接工厂，也就是说，你要往哪个消息服务器发：

    @Test
    public void testSendMsg() {
        SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), "rabbitConnectionFactory");
        rabbitTemplate.convertAndSend("exchange", "rkey.test", "test");
        SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());

        SimpleResourceHolder.bind(rabbitTemplate.getConnectionFactory(), "rabbitConnectionFactory1");
        rabbitTemplate.convertAndSend("exchange", "rkey.test1", "test1");
        SimpleResourceHolder.unbind(rabbitTemplate.getConnectionFactory());
    }

在发消息之前调用 SimpleResourceHolder.bind 绑定要使用的工厂，发完之后，调用 unbind 解除绑定。将上述代码封装为两个工具类，更好。

然后，有一个大坑在前面。。。如何收消息？

发消息要绑定连接工厂，指明往哪个消息服务器上发，收的时候，同样得指定要从哪个消息服务器上收。最开始没想到这点，以为只要指定队列名称就可以，如下：

    @RabbitListener(queues = "queue.test")
    public void receiveMsg(Message message) {
        String msg = new String(message.getBody());
        System.out.println(msg);
    }

然并卵，报了异常：

java.lang.IllegalStateException: Cannot determine target ConnectionFactory for lookup key [null]
    at org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.determineTargetConnectionFactory(AbstractRoutingConnectionFactory.java:116) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.AbstractRoutingConnectionFactory.createConnection(AbstractRoutingConnectionFactory.java:94) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils$1.createConnection(ConnectionFactoryUtils.java:80) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.doGetTransactionalResourceHolder(ConnectionFactoryUtils.java:130) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.getTransactionalResourceHolder(ConnectionFactoryUtils.java:67) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.start(BlockingQueueConsumer.java:456) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1158) ~[spring-rabbit-1.5.1.RELEASE.jar:na]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_45]

这个问题不好解决，查了很多资料都没用，比如这种方式：https://stackoverflow.com/questions/42784471/spring-amqp-mix-simpleroutingconnectionfactory-with-rabbitlistener  。

无奈之下，只能试着看看 Spring 的 AMQP 怎么实现，看看有没有解决的办法，最开始想的是继承 Spring 的某个类来实现。然而，看来看去，很是头大，没有结果。

最后无意间点到了 @RabbitListener 这个注解中，发现了有一个属性，瞬间感觉很兴奋，如下图：

看了下注释，这里可以指定一个 containerFactory，感觉可以试试。首先只有一个 containerFactory，那就加一个吧。为了看的比较清晰，我把第一次添加的注释去掉了，于是配置成了这样：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/rabbit
       http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
        
    
    <rabbit:connection-factory id="rabbitConnectionFactory" host="${rabbit.host}" port="${rabbit.port}" username="${rabbit.username}"
            password="${rabbit.password}" requested-heartbeat="30" virtual-host="${rabbit.vhost}" channel-cache-size="50"/>
    
    <rabbit:connection-factory id="rabbitConnectionFactory1" host="${rabbit.host1}" port="${rabbit.port1}" username="${rabbit.username1}" 
　　　　　　　password="${rabbit.password1}" requested-heartbeat="30" virtual-host="${rabbit.vhost1}" channel-cache-size="50"/>

    <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.SimpleRoutingConnectionFactory">
        <property name="targetConnectionFactories">
            <map>
                <entry key="rabbitConnectionFactory" value-ref="rabbitConnectionFactory"/>
                <entry key="rabbitConnectionFactory1" value-ref="rabbitConnectionFactory1"/>
            </map>
        </property>
    </bean>
    
    <rabbit:admin id="rabbitAdmin" connection-factory="rabbitConnectionFactory"/>
    
    <!-- 添加一个 rabbitAdmin-->
    <rabbit:admin id="rabbitAdmin1" connection-factory="rabbitConnectionFactory1"/>

    
    <!-- 把原有的 ContainerFactory 的连接工厂改为 rabbitConnectionFactory-->
    <bean id="rabbitListenerContainerFactory"
          class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="rabbitConnectionFactory"/>
        <property name="concurrentConsumers" value="16"/>
        <property name="maxConcurrentConsumers" value="50"/>
    </bean>
    
    <!-- 添加一个 ContainerFactory， 连接工厂为 rabbitConnectionFactory1-->
    <bean id="rabbitListenerContainerFactory1"
          class="org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory">
        <property name="connectionFactory" ref="rabbitConnectionFactory1"/>
        <property name="concurrentConsumers" value="16"/>
        <property name="maxConcurrentConsumers" value="50"/>
    </bean>

    <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="queue.test"/>
    <rabbit:queue durable="true" auto-delete="false" exclusive="false" name="queue.test1"/>
    
    <rabbit:direct-exchange name="exchange" auto-delete="false" durable="true">
        <rabbit:bindings>
            <rabbit:binding queue="queue.test" key="rkey.test"></rabbit:binding>
            <rabbit:binding queue="queue.test1" key="rkey.test1"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory" retry-template="retryTemplate" reply-timeout="60000"/>

    <bean id="retryTemplate" class="org.springframework.retry.support.RetryTemplate">
        <property name="backOffPolicy">
            <bean class="org.springframework.retry.backoff.ExponentialBackOffPolicy">
                <property name="initialInterval" value="500"/>
                <property name="multiplier" value="10.0"/>
                <property name="maxInterval" value="10000"/>
            </bean>
        </property>
    </bean>
</beans>

收消息的时候指定 container factory 即可：

    @RabbitListener(queues = "queue.test", containerFactory = "rabbitListenerContainerFactory")
    public void receiveMsg(Message message) {
        String msg = new String(message.getBody());
        System.out.println(msg);
    }

测试通过！

以上配置、解决办法是尝试过多次以后得出的，所以还是要有耐心，多尝试。

由于在网上没有找到解决办法，只有自己摸索着解决，如果大家有其他解决方案，欢迎留言讨论！