RabbitMQ笔记(四)-CachingConnectionFactory

通常我们使用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;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

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

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

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

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

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

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

于是配置两个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();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

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

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

rabbitTemplate.setUsePublisherConnection(true);
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());
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

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);
    }
}
1
2
3
4
5
6
7
8
9
10
11

欢迎关注微信交流