通常我们使用RabbitTemplate来进行简单的收发消息,而RabbitTemplate使用CachingConnectionFactory作为连接工厂,
CachingConnectionFactory
@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;
}
在一个应用里面同时存在消费者和生产者时,需要特别注意
文档建议使用一个具有相同选项的单独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();
}
将@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);
}
}