本文介绍在 AMQP 协议下，使用 Spring Boot 连接 RabbitMQ 实例时，如何配置连接隔离。 ## 准备工作 * 在 RabbitMQ 实例详情页中获取 AMQP 接入点。如需为实例开启公网访问，请参见[设置公网访问](https://docs.volcengine.com/docs/6451/70963)。 * 直接使用管理员用户进行生产和消费，可能会有信息泄露风险。您在进行生产和消费前，需要先在 RabbitMQ 集群管理 Web UI 上创建一个非 Admin 角色的新用户，并为新用户绑定 Virtual Host。操作步骤，请参见[绑定用户和 Vhost](https://docs.volcengine.com/docs/6451/1209469)。 * 提前在 RabbitMQ 集群管理 Web UI 上创建 Queue 和 Exchange。 ## 步骤一：添加依赖在`pom.xml`文件中添加以下依赖： ```XML org.springframework.boot spring-boot-starter-amqp 2.3.1.RELEASE org.springframework.boot spring-boot-starter-web 2.3.1.RELEASE ``` ## 步骤二：准备配置文件在`resources`路径下，创建配置文件`application.yml`。请正确填写 RabbitMQ 实例相关配置后，将以下内容填写到配置文件中。 ```YAML spring: rabbitmq: host: rbtmq-516***.rabbitmq.volces.com port: 5671 virtual-host: / username: doc-user password: user-password # 如果实例开启公网访问，需要配置如下信息。 ssl: enabled: true algorithm: TLSv1.2 verify-hostname: false validate-server-certificate: false ``` ## 步骤三：配置连接隔离选择以下任意一种策略实现连接隔离。使用独立 Publisher Connection 在`RabbitTemplate`中设置`setUsePublisherConnection(true)`，使消息发布使用独立的`publisher connection`，示例代码如下： ```Java @Bean public RabbitTemplate rabbitTemplate(CachingConnectionFactory cf) { RabbitTemplate template = new RabbitTemplate(cf); template.setUsePublisherConnection(true); // 使用独立的 publisher connection 发布消息 return template; } ``` 显式配置双 ConnectionFactory 分别配置消费端和生产端的`ConnectionFactory`，并使用`@Qualifier("consumerConnectionFactory")`和`@Qualifier("publisherConnectionFactory")`显式绑定`RabbitTemplate`和`SimpleRabbitListenerContainerFactory`。同时，可以通过`connectionNameStrategy`配置连接名称，方便在`Broker`侧通过`list_connections`区分不同连接。

注意

使用显式双ConnectionFactory时，需要为所有ConnectionFactory注入点配置对应的@Qualifier。如果存在未配置@Qualifier的@Autowired ConnectionFactory，Spring 会注入@Primary标记的连接工厂，可能导致连接隔离不生效。

## 步骤四：配置超时与确认参数选择以下任意一种方式配置超时与确认参数。配置文件配置 ```YAML spring: rabbitmq: connection-timeout: 30s publisher-confirm-type: correlated # none / simple / correlated，默认是 fire-and-forget，无法超时 publisher-returns: true cache: channel: size: 25 # channelCacheSize，请根据业务实际情况调整至合适值 checkout-timeout: 5s # channelCheckoutTimeout，大于 0 时 size 变成硬上限 ``` 代码配置 ```Java ConnectionFactory rabbitFactory = new ConnectionFactory(); rabbitFactory.setConnectionTimeout(30_000); CachingConnectionFactory cf = new CachingConnectionFactory(rabbitFactory); // publisher confirms：使用 ConfirmType，避免使用已废弃的 setPublisherConfirms(true) cf.setPublisherConfirmType(ConfirmType.CORRELATED); // NONE / SIMPLE / CORRELATED cf.setPublisherReturns(true); // 通常和 confirm 一起开启，用于处理不可路由消息 // 获取 Channel 的超时时间，单位为毫秒 cf.setChannelCheckoutTimeout(5000); // 获取不到 Channel 时快速失败，默认值 0 表示无限等待 ``` ## 步骤五：配置 BlockedListener 与业务熔断建议增加`BlockedListener`和业务熔断配置，避免 Tomcat 线程池被耗尽。 1. 在`publisherConnectionFactory`上添加事件监听，示例代码如下： ```Java ConnectionFactory publisherCf = cf.getPublisherConnectionFactory(); if (publisherCf != null) { publisherCf.addConnectionListener(connection -> connection.addBlockedListener(new BlockedListener() { @Override public void handleBlocked(String reason) { log.warn("Publisher 连接被 BLOCK，原因: {}", reason); } @Override public void handleUnblocked() { log.info("Publisher 连接已 UNBLOCK，恢复发布"); } }) ); } ``` 2. 根据`BlockedListener`状态，判断是否允许发送消息，如果不允许，应避免向 RabbitMQ 服务端发起请求，业务可进行降级处理（`fallback`），将消息存储到其他存储系统。

消息队列 RabbitMQ版