SpringBoot 整合RabbitMQ 之:CachingConnectionFactory 连接工厂
SpringBoot 整合RabbitMQ 消息队列服务过程中,代码程序连接 RabbitMQ 服务需要用到重要的类有:ConnectionFactory,Connection,Channel和QueueingConsumer。
1.1、RabbitMQ 连接创建过程
- 首先通过 new ConnectionFactory() 创建一个 ConnectionFactory;并设置此连接工厂的主机设置为localhost。
- 通过 ConnectionFactory 的 newConnection() 方法创建一个 Connection;
- newConnection 方法通过得到当前连接的地址及端口号来获得一个Address,通过 createFrameHandler 的方法来得到 FrameHandler;
- 再通过 new AMQConnection(this, frameHandler) 来得到Connection并启动。
1.2、ConnectionFactory的作用
- 利用工厂模式提升代码的重用性
- 封装注册数据库的驱动和获得数据库的连接
- 利用配置文件减少硬编码,便于维护
二、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);
}
}