rabbitmq的多交换机(扇出为例)监听实现

1. 配置文件，微服务或者单体都可以在yml里配置，也可以写死在java文件.

spring:rabbitmq:addresses: 127.0.0.1:5672username: guestpassword: guestvirtual-host: /connection-timeout: 15000#配置rabbitMq启用开关enable: truelistener:simple:acknowledge-mode: manualconcurrency: 15max-concurrency: 20

3. 监听处理文件

package jzy.ziyuan.util.rabbitmq;import cn.hutool.json.JSONObject;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import jnpf.util.JsonUtil;
import jzy.model.mom.EventMessageMom;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.io.IOException;@Component
@ConditionalOnProperty(name = "spring.rabbitmq.enable", havingValue = "true", matchIfMissing = true)
public class MultiExchangeMessageListener {private static final Logger logger = LoggerFactory.getLogger(MultiExchangeMessageListener.class);@Autowiredprivate IdempotentMessageProcessor idempotentProcessor;@Autowiredprivate RabbitTemplate rabbitTemplate;/*** 监听来自多个交换机的消息*/// 声明 Direct 交换机、队列，并绑定路由键@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "order.queue", durable = "true"),    // 队列exchange = @Exchange(name = "20250917test1", type = ExchangeTypes.FANOUT) // 交换机))@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "user.queue", durable = "true"),    @RabbitListener(queues = "#{tempQueue.name}") // 引用临时队列public void handleMessage(Message message) throws IOException {String messageId = message.getMessageProperties().getMessageId();String exchange = message.getMessageProperties().getReceivedExchange();long deliveryTag = message.getMessageProperties().getDeliveryTag();try {// 1. 幂等性检查if (idempotentProcessor.isProcessed(messageId)) {logger.info("消息已处理，跳过重复消费: messageId={}, exchange={}", messageId, exchange);// 手动确认消息rabbitTemplate.getConnectionFactory().createConnection().createChannel(false).basicAck(deliveryTag, false);return;}// 2. 处理消息内容String messageBody = new String(message.getBody(), "UTF-8");String processedMessage = replaceNbspWithSpace(messageBody);// 3. 解析消息JSONObject jsonObject = JsonUtil.getJsonToBean(processedMessage, JSONObject.class);EventMessageMom eventMessage = JsonUtil.getJsonToBean(jsonObject, EventMessageMom.class);logger.info("接收来自交换机的消息: exchange={}, messageId={}, content={}",exchange, messageId, eventMessage);switch (exchange) {default:break;}// 4. 执行业务逻辑// 5. 标记为已处理idempotentProcessor.markAsProcessed(messageId);// 6. 手动确认消息rabbitTemplate.getConnectionFactory().createConnection().createChannel(false).basicAck(deliveryTag, false);logger.info("消息处理成功: messageId={}, exchange={}", messageId, exchange);} catch (Exception e) {logger.error("消息处理失败: messageId={}, exchange={}", messageId, exchange, e);// 业务失败，拒绝消息并放入死信队列rabbitTemplate.getConnectionFactory().createConnection().createChannel(false).basicNack(deliveryTag, false, false); // 不重新入队}}/*** 过滤特殊符号*/public static String replaceNbspWithSpace(String str) {if (str == null) {return null;}return str.replaceAll("\\u00A0", " ");}
}

5. rabbitmq配置文件

package jzy.ziyuan.util.rabbitmq;import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class MultiExchangeRabbitConfig {// 交换机名称列表 - 可根据实际需求扩展public static final String[] EXCHANGE_NAMES = {"exchange.20250918logs","exchange.20250918events","exchange.20250918notifications"};// 交换机类型public static final String EXCHANGE_TYPE = "fanout";// 死信交换机和队列public static final String DLX_EXCHANGE = "20250918exchange.dlx";public static final String DLX_QUEUE = "20250918queue.dlx";/*** 创建死信交换机*/@Beanpublic DirectExchange dlxExchange() {return ExchangeBuilder.directExchange(DLX_EXCHANGE).durable(true).build();}/*** 创建死信队列*/@Beanpublic Queue dlxQueue() {return QueueBuilder.durable(DLX_QUEUE).build();}/*** 绑定死信队列到死信交换机*/@Beanpublic Binding dlxBinding() {return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("dlx.routing.key");}/*** 创建临时队列 - 非持久化、排他、自动删除*/@Beanpublic Queue tempQueue() {// 配置死信参数Map<String, Object> args = new HashMap<>();args.put("x-dead-letter-exchange", DLX_EXCHANGE);args.put("x-dead-letter-routing-key", "dlx.routing.key");return QueueBuilder.nonDurable()    // 非持久化.exclusive()                // 排他队列.autoDelete()               // 自动删除.withArguments(args).build();}/*** 创建多个交换机并绑定到临时队列*/@Beanpublic Declarables exchangeBindings() {Declarables declarables = new Declarables();// 创建所有交换机并绑定到临时队列for (String exchangeName : EXCHANGE_NAMES) {// 创建交换机FanoutExchange exchange = ExchangeBuilder.fanoutExchange(exchangeName).durable(true).build();declarables.getDeclarables().add(exchange);// 绑定交换机到临时队列Binding binding = BindingBuilder.bind(tempQueue()).to(exchange);declarables.getDeclarables().add(binding);}return declarables;}/*** 配置监听容器工厂 - 手动ACK和QoS设置*/@Beanpublic RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConnectionFactory(connectionFactory);factory.setAcknowledgeMode(AcknowledgeMode.MANUAL); // 手动确认factory.setPrefetchCount(10); // QoS设置factory.setConcurrentConsumers(3); // 并发消费者数量factory.setMaxConcurrentConsumers(5); // 最大并发消费者数量// 配置自动重连机制factory.setRecoveryInterval(5000L); // 重连间隔5秒return factory;}
}