RabbitMQ笔记(四)-CachingConnectionFactory

电玩女神 2022-03-27 11:26 748阅读 0赞

通常我们使用RabbitTemplate来进行简单的收发消息，而RabbitTemplate使用CachingConnectionFactory作为连接工厂，

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;
    }

#### 通常的教程这样配置，但是特别注意到官网有一段这样的提示 ####

在一个应用里面同时存在消费者和生产者时，需要特别注意

![Image 1][]

文档建议使用一个具有相同选项的单独CachingConnectionFactory实例—一个用于生产者，一个用于消费者。

这是为了避免消费者由于生产者阻塞而阻塞

这里可以做一个测试，首先将RabbitMQ的内存水位调低，产生内存报警

![Image 1][]

再发送生产者的消息时，会发现产生了阻塞，同时添加一个监听者，这条指令同样也会发送阻塞

于是配置两个CachingConnectionFactory

private CachingConnectionFactory getCachingConnectionFactory() {
            CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
    
            cachingConnectionFactory.setAddresses(rabbitProperties.getAddresses());
            cachingConnectionFactory.setUsername(rabbitProperties.getUsername());
            cachingConnectionFactory.setPassword(rabbitProperties.getPassword());
            cachingConnectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());
            cachingConnectionFactory.setPublisherConfirms(rabbitProperties.isPublisherConfirms());
            cachingConnectionFactory.setPublisherReturns(rabbitProperties.isPublisherReturns());
    
    
            cachingConnectionFactory.addChannelListener(rabbitChannelListener);
            cachingConnectionFactory.addConnectionListener(rabbitConnectionListener);
            cachingConnectionFactory.setRecoveryListener(rabbitRecoveryListener);
    
    
            return cachingConnectionFactory;
        }
    
        @Bean("test-consumer-connection-factory")
        public CachingConnectionFactory consumerCachingConnectionFactory() {
    
            return getCachingConnectionFactory();
    
        }
    
        @Bean
        @Primary
        public CachingConnectionFactory cachingConnectionFactory() {
    
            return getCachingConnectionFactory();
        }

将@Bean(“test-consumer-connection-factory”) 用于消费者  
则在发送阻塞之后，消费者的通道仍然是畅通的

当然由于使用RabbitTemplate，也可以在RabbitTemplate配置

rabbitTemplate.setUsePublisherConnection(true);

#### 这里有三个监听器 ####

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());
            }
        }
    }

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);
        }
    }

RecoveryListener  
监听自动重连的情况，这个listener没有测试出在什么场景会出现

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);
        }
    }

欢迎关注微信交流  
![在这里插入图片描述][70]

[Image 1]: 
[70]: /images/20220327/df84b820cc6040edb05187d772fdd71d.png