SpringBoot 整合RabbitMQ 之：CachingConnectionFactory 连接工厂

SpringBoot 整合RabbitMQ 消息队列服务过程中，代码程序连接 RabbitMQ 服务需要用到重要的类有：ConnectionFactory，Connection，Channel和QueueingConsumer。

1.1、RabbitMQ 连接创建过程

1. 首先通过 new ConnectionFactory() 创建一个 ConnectionFactory；并设置此连接工厂的主机设置为localhost。
2. 通过 ConnectionFactory 的 newConnection() 方法创建一个 Connection;
3. newConnection 方法通过得到当前连接的地址及端口号来获得一个Address，通过 createFrameHandler 的方法来得到 FrameHandler;
4. 再通过 new AMQConnection(this, frameHandler) 来得到Connection并启动。

1.2、ConnectionFactory的作用

1. 利用工厂模式提升代码的重用性
2. 封装注册数据库的驱动和获得数据库的连接
3. 利用配置文件减少硬编码，便于维护

二、CachingConnectionFactory 连接工厂使用

通常我们使用 RabbitTemplate 来进行简单的收发消息，而 RabbitTemplate 使用 CachingConnectionFactory 作为连接工厂。
SpringBoot 配置类中设置 CachingConnectionFactory 类的Bean 代码。

@Bean
public CachingConnectionFactory cachingConnectionFactory(){
    CachingConnectionFactory factory = new CachingConnectionFactory();

    factory.setAddresses(rabbitProperties.getAddresses());
    factory.setUsername(rabbitProperties.getUsername());
    factory.setPassword(rabbitProperties.getPassword());
    factory.setVirtualHost(rabbitProperties.getVirtualHost());
    factory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
    factory.setPublisherReturns(rabbitProperties.isPublisherReturns());

    factory.addChannelListener(rabbitChannelListener);
    factory.addConnectionListener(rabbitConnectionListener);
    factory.setRecoveryListener(rabbitRecoveryListener);

    return factory;
}

2.1、注意点

如果生产者和消费者使用同一个 CachingConnectionFactory；则两者会使用同一个 Connection和 Channel 实例；如果某个客户端阻塞了，则会影响到另外的客户端。
所以，应该使用分别创建 CachingConnectionFactory。

@Bean("consumer-connection-factory")
public CachingConnectionFactory consumerCachingConnectionFactory() {
    return getCachingConnectionFactory();
}

@Bean
@Primary
public CachingConnectionFactory cachingConnectionFactory() {
    return getCachingConnectionFactory();
}

这样的话，将@Bean(“consumer-connection-factory”) 用于消费者；就可以与生产者区分开了。当然也可以使用 rabbitTemplate.setUsePublisherConnection(true); 实现。

三、CachingConnectionFactory 中的监听器

CachingConnectionFactory 中有三个监听器：ChannelListener、ConnectionListener和RecoveryListener 监听器。

3.1、ChannelListener 监听通道的创建和销毁

@Service
public class RabbitChannelListener implements ChannelListener {
    @Override
    public void onCreate(Channel channel, boolean b) {
        log.info("==onCreate channel: {}, transactional: {}", channel, b);
    }
    @Override
    public void onShutDown(ShutdownSignalException signal){
    // 可根据isHardError判断是channel断开还是connection断开
        if(signal.isHardError()){
            AMQImpl.Connection.Close close = (AMQImpl.Connection.Close) signal.getReason();
            log.warn("==Connection onShutDown replyCode: {}, methodId: {}, classId: {}, replyText: {}",
                    close.getReplyCode(), close.getMethodId(), close.getClassId(), close.getReplyText());
        }else {
            AMQImpl.Channel.Close close = (AMQImpl.Channel.Close) signal.getReason();
            log.warn("==Channel onShutDown replyCode: {}, methodId: {}, classId: {}, replyText: {}",
                    close.getReplyCode(), close.getMethodId(), close.getClassId(), close.getReplyText());
        }
    }
}

3.2、ConnectionListener 监听连接的创建和关闭

public class RabbitConnectionListener implements ConnectionListener {
    @Override
    public void onCreate(Connection connection) {
        log.info("==onCreate: {}", connection);
    }
    @Override
    public void onClose(Connection connection) {
        log.info("==onClose: {}", connection);
    }
    @Override
    public void onShutDown(ShutdownSignalException signal) {
        log.info("==onShutDown: {}", signal);
    }
}

3.3、RecoveryListener 监听自动重连的情况

public class RabbitRecoveryListener implements RecoveryListener {
    @Override
    public void handleRecovery(Recoverable recoverable) {
        log.info("==handleRecovery: {}", recoverable);
    }
    @Override
    public void handleRecoveryStarted(Recoverable recoverable) {
        log.info("==handleRecoveryStarted: {}", recoverable);
    }
}