当前位置：首页 > news >正文

【中间件】RabbitMQ消息队列实战：从入门到精通

news 2026/5/25 2:17:28

【中间件】RabbitMQ消息队列实战：从入门到精通

引言

RabbitMQ是采用Erlang语言实现的高级消息队列协议（AMQP）的开源消息代理软件，在企业级应用中扮演着至关重要的角色。它提供可靠的消息传递、灵活的路由、集群高可用、扩展性强等特性，被广泛应用于异步处理、系统解耦、流量削峰等场景。本文将从基础概念出发，详细讲解RabbitMQ的安装配置、核心概念、交换机类型、消息确认机制、集群部署以及最佳实践，帮助读者全面掌握RabbitMQ消息队列技术。

一、RabbitMQ概述

1.1 什么是消息队列

消息队列（Message Queue）是一种进程间通信或同一进程的不同线程间的通信方式。它的核心作用是解耦生产者和消费者，使得消息发送者和接收者不需要同时在线，也不需要同时处理消息。常见的消息队列产品包括RabbitMQ、Apache Kafka、Redis Streams、ActiveMQ等。

1.2 RabbitMQ核心特性

可靠性：支持消息持久化、发布确认、消费确认机制
灵活的路由：通过交换机（Exchange）和绑定（Binding）实现复杂路由
集群高可用：支持镜像队列，实现主从复制
多协议支持：AMQP、MQTT、STOMP、HTTP
丰富的客户端：支持所有主流编程语言
可视化管理：提供Web管理界面
插件扩展：支持多种插件扩展功能

1.3 AMQP协议模型

┌─────────────────────────────────────────────────────────────────┐ │ RabbitMQ Architecture │ │ │ │ Producer │ │ │ │ │ ▼ │ │ ┌─────────┐ │ │ │ Exchange │ ◄──── Binding ────┐ │ │ └────┬────┘ │ │ │ │ │ │ │ ├───────────────────┬───┴────────────────────────┐ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ Queue │ │ Queue │ │ Queue │ │ │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ │ │ │ │ └───────────────────┼────────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────┐ │ │ │Consumer │ │ │ └─────────┘ │ └─────────────────────────────────────────────────────────────────┘

二、核心概念详解

2.1 生产者、消费者与消息

# Python pika客户端 - 生产者 import pika import json from datetime import datetime class OrderProducer: def __init__(self, host='localhost', port=5672): self.connection = pika.BlockingConnection( pika.ConnectionParameters(host=host, port=port) ) self.channel = self.connection.channel() def publish_order_created(self, order_data): """发布订单创建事件""" message = { 'event_type': 'ORDER_CREATED', 'order_id': order_data['order_id'], 'user_id': order_data['user_id'], 'total_amount': order_data['total_amount'], 'items': order_data['items'], 'timestamp': datetime.now().isoformat() } # 发布到交换机 self.channel.basic_publish( exchange='order.events', routing_key='order.created', body=json.dumps(message), properties=pika.BasicProperties( delivery_mode=2, # 持久化消息 content_type='application/json', message_id=order_data['order_id'], timestamp=int(datetime.now().timestamp()) ) ) print(f"Published order created event: {order_data['order_id']}") def publish_order_cancelled(self, order_id, reason): """发布订单取消事件""" message = { 'event_type': 'ORDER_CANCELLED', 'order_id': order_id, 'reason': reason, 'timestamp': datetime.now().isoformat() } self.channel.basic_publish( exchange='order.events', routing_key='order.cancelled', body=json.dumps(message) ) def close(self): self.connection.close() # 消费者 class OrderConsumer: def __init__(self, host='localhost', port=5672, queue='order.notifications'): self.connection = pika.BlockingConnection( pika.ConnectionParameters(host=host, port=port) ) self.channel = self.connection.channel() self.queue = queue # 设置QoS - 预取消息数量 self.channel.basic_qos(prefetch_count=10) def callback(self, ch, method, properties, body): """消息处理回调""" try: message = json.loads(body) print(f"Received message: {message}") # 处理消息 self.process_message(message) # 手动确认消息 ch.basic_ack(delivery_tag=method.delivery_tag) except json.JSONDecodeError as e: print(f"JSON decode error: {e}") # 拒绝消息，不重新入队 ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) except Exception as e: print(f"Processing error: {e}") # 处理失败，重新入队 ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True) def process_message(self, message): """处理不同类型的消息""" event_type = message.get('event_type') if event_type == 'ORDER_CREATED': self.handle_order_created(message) elif event_type == 'ORDER_CANCELLED': self.handle_order_cancelled(message) else: print(f"Unknown event type: {event_type}") def handle_order_created(self, message): """处理订单创建""" # 发送通知、更新统计等 pass def handle_order_cancelled(self, message): """处理订单取消""" # 释放库存、发送通知等 pass def start_consuming(self): """开始消费""" self.channel.basic_consume( queue=self.queue, on_message_callback=self.callback ) print(f"Started consuming from queue: {self.queue}") self.channel.start_consuming() def stop_consuming(self): """停止消费""" self.channel.stop_consuming() self.connection.close()

2.2 交换机类型

RabbitMQ提供了四种交换机类型，每种类型有不同的路由规则：

2.2.1 Direct交换机

# Direct交换机 - 点对点路由 # 路由规则：根据routing key精确匹配 # 声明交换机 channel.exchange_declare( exchange='order.direct', exchange_type='direct', durable=True ) # 声明队列 channel.queue_declare(queue='order.email.queue', durable=True) channel.queue_declare(queue='order.sms.queue', durable=True) # 绑定队列到交换机 channel.queue_bind( exchange='order.direct', queue='order.email.queue', routing_key='order.email' # 精确匹配 ) channel.queue_bind( exchange='order.direct', queue='order.sms.queue', routing_key='order.sms' ) # 发送消息 channel.basic_publish( exchange='order.direct', routing_key='order.email', # 只有email队列收到 body='Order notification via email' ) channel.basic_publish( exchange='order.direct', routing_key='order.sms', # 只有sms队列收到 body='Order notification via SMS' )

2.2.2 Fanout交换机

# Fanout交换机 - 广播到所有绑定的队列 # 路由规则：忽略routing key，发送到所有绑定的队列 # 声明fanout交换机 channel.exchange_declare( exchange='notifications.fanout', exchange_type='fanout', durable=True ) # 声明多个队列 queues = ['email.notifications', 'sms.notifications', 'push.notifications'] for queue_name in queues: channel.queue_declare(queue=queue_name, durable=True) channel.queue_bind( exchange='notifications.fanout', queue=queue_name ) # 发布消息，所有队列都会收到 channel.basic_publish( exchange='notifications.fanout', routing_key='', # fanout会忽略routing key body='This message goes to all notification queues' )

2.2.3 Topic交换机

# Topic交换机 - 通配符路由 # 路由规则：根据routing key模式匹配 # * 匹配一个单词 # # 匹配零个或多个单词 # 声明topic交换机 channel.exchange_declare( exchange='logs.topic', exchange_type='topic', durable=True ) # 绑定队列，使用通配符 bindings = [ ('log.system.queue', 'log.system.*'), # system.* 匹配 system.error, system.warning ('log.application.queue', 'log.*.app'), # *.*.app 匹配 any.any.app ('log.all.queue', 'log.#'), # # 匹配所有log开头的 ('log.critical.queue', 'log.system.critical') # 精确匹配 ] for queue_name, routing_pattern in bindings: channel.queue_declare(queue=queue_name, durable=True) channel.queue_bind( exchange='logs.topic', queue=queue_name, routing_key=routing_pattern ) # 发送消息 routing_keys = [ 'log.system.error', 'log.system.warning', 'log.database.query', 'log.payment.app.success', 'log.payment.app.failure', 'log.audit.security' ] for key in routing_keys: channel.basic_publish( exchange='logs.topic', routing_key=key, body=f'Log message for {key}' )

2.2.4 Headers交换机

# Headers交换机 - 根据消息头匹配 # 路由规则：根据headers属性匹配，比routing key更灵活 channel.exchange_declare( exchange='data.headers', exchange_type='headers', durable=True ) # 绑定队列，指定headers channel.queue_declare(queue='video.queue', durable=True) channel.queue_bind( exchange='data.headers', queue='video.queue', arguments={ 'x-match': 'all', # all=所有header都匹配, any=任一匹配 'content-type': 'video/*', 'priority': 'high' } ) channel.queue_declare(queue='image.queue', durable=True) channel.queue_bind( exchange='data.headers', queue='image.queue', arguments={ 'x-match': 'any', 'content-type': 'image/*', 'compress': 'true' } ) # 发送消息 channel.basic_publish( exchange='data.headers', routing_key='', # headers交换机忽略routing key body=video_data, properties=pika.BasicProperties( headers={ 'content-type': 'video/mp4', 'priority': 'high', 'encoding': 'h264' } ) )

三、消息确认与可靠性

3.1 消息持久化

# 消息持久化配置 class ReliableProducer: def __init__(self): self.connection = pika.BlockingConnection( pika.ConnectionParameters( host='localhost', heartbeat=600, blocked_connection_timeout=300 ) ) self.channel = self.connection.channel() # 开启发布确认 self.channel.confirm_delivery() def publish_with_confirmation(self, message): """发布消息并等待确认""" try: self.channel.basic_publish( exchange='reliable.exchange', routing_key='reliable.route', body=message, properties=pika.BasicProperties( delivery_mode=2, # 持久化消息到磁盘 content_type='application/json' ), mandatory=True # 消息必须路由到队列，否则返回确认 ) print("Message published and confirmed") except pika.exceptions.UnroutableMetadata: print("Message could not be routed") # 处理无法路由的消息 self.handle_unroutable_message(message) except pika.exceptions.NackError: print("Message was Nacked by broker") def handle_unroutable_message(self, message): """处理无法路由的消息""" # 可以记录到数据库或发送到死信队列 pass

3.2 消费确认机制

# 手动确认与重试机制 class RetryConsumer: def __init__(self): self.max_retries = 3 self.retry_delay = 5 # 秒 def process_with_retry(self, ch, method, properties, body): """带重试机制的消息处理""" headers = properties.headers or {} retry_count = headers.get('x-retry-count', 0) try: # 处理消息 self.process_message(body) # 成功确认 ch.basic_ack(delivery_tag=method.delivery_tag) except TemporaryException as e: # 临时错误，稍后重试 if retry_count < self.max_retries: # 重新发布消息，增加重试计数 self.republish_with_retry(body, retry_count + 1) ch.basic_ack(delivery_tag=method.delivery_tag) else: # 超过最大重试次数，发送到死信队列 self.send_to_dlq(body, f"Max retries exceeded: {e}") ch.basic_ack(delivery_tag=method.delivery_tag) except PermanentException as e: # 永久错误，直接发送到死信队列 self.send_to_dlq(body, f"Permanent error: {e}") ch.basic_ack(delivery_tag=method.delivery_tag) def republish_with_retry(self, body, retry_count): """重新发布带重试计数""" properties = pika.BasicProperties( headers={'x-retry-count': retry_count}, delivery_mode=2 ) # 延迟重新发布 time.sleep(self.retry_delay * retry_count) self.channel.basic_publish( exchange='retry.exchange', routing_key='retry.route', body=body, properties=properties ) def send_to_dlq(self, body, error_reason): """发送到死信队列""" properties = pika.BasicProperties( headers={ 'x-error-reason': error_reason, 'x-original-body': body.decode('utf-8') }, delivery_mode=2 ) self.channel.basic_publish( exchange='dlx.exchange', routing_key='dlq', body=body, properties=properties )

3.3 死信队列配置

# 死信交换机和队列配置 def setup_dlx(channel): """配置死信交换机和队列""" # 声明死信交换机 channel.exchange_declare( exchange='dlx.exchange', exchange_type='direct', durable=True ) # 声明死信队列 channel.queue_declare( queue='dlq.orders', durable=True ) # 绑定死信队列 channel.queue_bind( exchange='dlx.exchange', queue='dlq.orders', routing_key='dlq.order' ) # 在主队列上配置死信交换机参数 channel.queue_declare( queue='orders', durable=True, arguments={ 'x-dead-letter-exchange': 'dlx.exchange', 'x-dead-letter-routing-key': 'dlq.order', 'x-message-ttl': 86400000, # 消息24小时过期 'x-max-length': 100000 # 队列最大消息数 } )

# Docker Compose配置RabbitMQ version: '3.8' services: rabbitmq: image: rabbitmq:3.12-management container_name: rabbitmq ports: - "5672:5672" # AMQP协议端口 - "15672:15672" # 管理界面端口 environment: - RABBITMQ_DEFAULT_USER=admin - RABBITMQ_DEFAULT_PASS=admin123 - RABBITMQ_DEFAULT_VHOST=/ volumes: - rabbitmq_data:/var/lib/rabbitmq - ./rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro healthcheck: test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"] interval: 30s timeout: 10s retries: 5 volumes: rabbitmq_data:

四、集群与高可用

4.1 集群配置

# rabbitmq.conf 集群配置 # 节点名称 node.name = rabbit@node1 # 集群配置 cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config cluster_formation.classic_config.nodes.1 = rabbit@node1 cluster_formation.classic_config.nodes.2 = rabbit@node2 cluster_formation.classic_config.nodes.3 = rabbit@node3 # 镜像队列策略 mirroring_sync_batch_size = 1 # 负载均衡 load_definitions = /etc/rabbitmq/definitions.json

# 集群管理命令 # 查看集群状态 rabbitmqctl cluster_status # 加入集群 rabbitmqctl join_cluster rabbit@node1 # 离开集群 rabbitmqctl leave_cluster # 变更节点类型 rabbitmqctl change_cluster_node_type disc rabbitmqctl change_cluster_node_type ram # 设置镜像策略 rabbitmqctl set_policy ha-all "^orders\." '{"ha-mode":"all","ha-sync-mode":"automatic"}'

4.2 镜像队列配置

# 镜像队列策略 """ ha-mode参数说明: - all: 队列镜像到所有节点 - exactly: 镜像到指定数量节点 - nodes: 镜像到指定节点列表 ha-sync-mode参数说明: - manual: 手动同步 - automatic: 自动同步 """ # 设置全局镜像策略 # rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}' # 设置特定队列的镜像策略 channel.queue_declare( queue='orders', durable=True, arguments={ 'x-queue-type': 'quorum', # 使用Quorum队列 'x-quorum-initial-group-size': 3 # Quorum队列副本数 } )

4.3 负载均衡配置

# Nginx TCP负载均衡配置 stream { upstream rabbitmq_cluster { least_conn; server node1:5672 weight=5; server node2:5672 weight=5; server node3:5672 weight=5; } server { listen 5672; proxy_pass rabbitmq_cluster; proxy_connect_timeout 1s; } }

五、Spring Boot集成

5.1 配置与依赖

<!-- pom.xml --> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> </dependency> </dependencies>

# application.yml spring: rabbitmq: host: localhost port: 5672 username: admin password: admin123 virtual-host: / # 连接池配置 pool: enabled: true max-size: 10 min-size: 2 # publisher确认 publisher-confirm-type: correlated publisher-returns: true # consumer配置 listener: simple: acknowledge-mode: manual prefetch: 10 retry: enabled: true initial-interval: 1000 max-attempts: 3 max-interval: 10000 multiplier: 2.0

5.2 RabbitTemplate使用

// RabbitTemplate消息发送 @Service public class OrderMessagePublisher { @Autowired private RabbitTemplate rabbitTemplate; // 发送订单创建事件 public void publishOrderCreated(Order order) { Map<String, Object> message = new HashMap<>(); message.put("orderId", order.getId()); message.put("userId", order.getUserId()); message.put("totalAmount", order.getTotalAmount()); message.put("items", order.getItems()); message.put("timestamp", LocalDateTime.now()); rabbitTemplate.convertAndSend( "order.exchange", "order.created", message, msg -> { msg.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); msg.getMessageProperties().setContentType("application/json"); msg.getMessageProperties().setMessageId(order.getId()); return msg; } ); } // 发送延迟消息（使用延迟插件） public void publishDelayMessage(Object message, long delayMs) { rabbitTemplate.convertAndSend( "delayed.exchange", "delayed.route", message, msg -> { msg.getMessageProperties().setDelay((int) delayMs); return msg; } ); } }

5.3 消息监听

// 消息监听器 @Component public class OrderMessageListener { private static final Logger log = LoggerFactory.getLogger(OrderMessageListener.class); @Autowired private OrderService orderService; @RabbitListener( queues = "order.notification.queue", concurrency = "3-10", // 并发消费者数量 ackMode = "MANUAL" ) public void handleOrderNotification( Map<String, Object> message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) { try { String orderId = (String) message.get("orderId"); log.info("Received order notification: {}", orderId); // 处理消息 orderService.sendNotification(message); // 确认消息 channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error("Error processing message", e); try { // 拒绝消息，重新入队 channel.basicNack(deliveryTag, false, true); } catch (IOException ioException) { log.error("Failed to nack message", ioException); } } } // 处理死信消息 @RabbitListener(queues = "dlq.orders") public void handleDeadLetter(Message message, Channel channel) { log.error("Received dead letter message: {}", new String(message.getBody())); // 记录日志、发送告警等 } }

5.4 交换机和队列声明

// RabbitMQ配置类 @Configuration public class RabbitMQConfig { // 交换机 @Bean public DirectExchange orderExchange() { return ExchangeBuilder .directExchange("order.exchange") .durable(true) .build(); } @Bean public FanoutExchange notificationExchange() { return ExchangeBuilder .fanoutExchange("notification.exchange") .durable(true) .build(); } @Bean public TopicExchange delayedExchange() { return ExchangeBuilder .topicExchange("delayed.exchange") .durable(true) .build(); } // 队列 @Bean public Queue orderNotificationQueue() { return QueueBuilder .durable("order.notification.queue") .withArgument("x-dead-letter-exchange", "dlx.exchange") .withArgument("x-dead-letter-routing-key", "dlq.order") .withArgument("x-message-ttl", 86400000) .build(); } @Bean public Queue orderProcessQueue() { return QueueBuilder .durable("order.process.queue") .quorum() // 使用Quorum队列 .build(); } @Bean public Queue dlqQueue() { return QueueBuilder .durable("dlq.orders") .build(); } // 绑定 @Bean public Binding orderNotificationBinding() { return BindingBuilder .bind(orderNotificationQueue()) .to(orderExchange()) .with("order.notification"); } @Bean public Binding orderProcessBinding() { return BindingBuilder .bind(orderProcessQueue()) .to(orderExchange()) .with("order.process"); } @Bean public Binding dlqBinding() { return BindingBuilder .bind(dlqQueue()) .to(ExchangeBuilder.directExchange("dlx.exchange").build()) .with("dlq.order"); } }

六、实战案例：订单系统

6.1 系统架构

┌─────────────────────────────────────────────────────────────────┐ │ Order System Architecture │ │ │ │ ┌─────────┐ ┌──────────────────────────────────────────┐ │ │ │ API │────►│ order.exchange (topic) │ │ │ │ Gateway │ └──────────────────┬───────────────────────┘ │ │ └─────────┘ │ │ │ │ │ │ │ ▼ ▼ │ │ ┌─────────┐ ┌──────────────────────────────────────────┐ │ │ │ Order │ │ Queues │ │ │ │ Service │ │ ┌────────────┐ ┌────────────┐ │ │ │ └────┬────┘ │ │notification│ │ process │ │ │ │ │ │ │ queue │ │ queue │ │ │ │ │ │ └─────┬──────┘ └──────┬─────┘ │ │ │ │ └─────────┼───────────────┼───────────────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────┐ ┌──────────────────────────────────────────┐ │ │ │ MySQL │ │ Consumers │ │ │ │ │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ └─────────┘ │ │ Email │ │ SMS │ │Inventory│ │ │ │ │ │ Service │ │ Service │ │ Service │ │ │ │ │ └─────────┘ └─────────┘ └─────────┘ │ │ │ └──────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────┘

6.2 完整实现

// 订单事件发布服务 @Service @Slf4j public class OrderEventPublisher { @Autowired private RabbitTemplate rabbitTemplate; public void publishOrderCreated(Order order) { OrderCreatedEvent event = OrderCreatedEvent.builder() .orderId(order.getId()) .userId(order.getUserId()) .items(order.getItems()) .totalAmount(order.getTotalAmount()) .paymentMethod(order.getPaymentMethod()) .shippingAddress(order.getShippingAddress()) .createdAt(LocalDateTime.now()) .build(); Map<String, Object> message = new HashMap<>(); message.put("eventType", "ORDER_CREATED"); message.put("orderId", order.getId()); message.put("data", event); rabbitTemplate.convertAndSend( "order.exchange", "order.created", message ); log.info("Published ORDER_CREATED event for order: {}", order.getId()); } public void publishOrderPaid(Order order) { Map<String, Object> message = new HashMap<>(); message.put("eventType", "ORDER_PAID"); message.put("orderId", order.getId()); message.put("paidAt", LocalDateTime.now()); rabbitTemplate.convertAndSend( "order.exchange", "order.paid", message ); } public void publishOrderCancelled(Order order, String reason) { Map<String, Object> message = new HashMap<>(); message.put("eventType", "ORDER_CANCELLED"); message.put("orderId", order.getId()); message.put("reason", reason); message.put("cancelledAt", LocalDateTime.now()); rabbitTemplate.convertAndSend( "order.exchange", "order.cancelled", message ); } } // 订单通知消费者 @Service @Slf4j public class OrderNotificationConsumer { @Autowired private EmailService emailService; @Autowired private SmsService smsService; @RabbitListener(queues = "order.notification.queue") public void handleNotification( Map<String, Object> message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) { try { String eventType = (String) message.get("eventType"); String orderId = (String) message.get("orderId"); log.info("Processing notification for order: {}, event: {}", orderId, eventType); switch (eventType) { case "ORDER_CREATED": handleOrderCreated(message); break; case "ORDER_PAID": handleOrderPaid(message); break; case "ORDER_CANCELLED": handleOrderCancelled(message); break; default: log.warn("Unknown event type: {}", eventType); } channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error("Error processing notification", e); try { channel.basicNack(deliveryTag, false, true); } catch (IOException ex) { log.error("Failed to nack message", ex); } } } private void handleOrderCreated(Map<String, Object> message) { @SuppressWarnings("unchecked") Map<String, Object> data = (Map<String, Object>) message.get("data"); String email = (String) data.get("customerEmail"); String orderId = (String) message.get("orderId"); emailService.sendOrderConfirmation(email, orderId); } private void handleOrderPaid(Map<String, Object> message) { // 处理支付成功通知 } private void handleOrderCancelled(Map<String, Object> message) { // 处理订单取消通知 } } // 库存服务消费者 @Service @Slf4j public class InventoryConsumer { @Autowired private InventoryService inventoryService; @RabbitListener(queues = "order.process.queue") public void handleInventory( Map<String, Object> message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) { String eventType = (String) message.get("eventType"); try { if ("ORDER_CREATED".equals(eventType)) { @SuppressWarnings("unchecked") Map<String, Object> data = (Map<String, Object>) message.get("data"); @SuppressWarnings("unchecked") List<Map<String, Object>> items = (List<Map<String, Object>>) data.get("items"); inventoryService.reserveStock(items); } else if ("ORDER_CANCELLED".equals(eventType)) { String orderId = (String) message.get("orderId"); inventoryService.releaseStock(orderId); } channel.basicAck(deliveryTag, false); } catch (InsufficientStockException e) { log.error("Insufficient stock for order", e); // 拒绝消息，发送到死信队列 try { channel.basicNack(deliveryTag, false, false); } catch (IOException ex) { log.error("Failed to nack message", ex); } } catch (Exception e) { log.error("Error processing inventory", e); try { channel.basicNack(deliveryTag, false, true); } catch (IOException ex) { log.error("Failed to nack message", ex); } } } }

七、性能优化与最佳实践

7.1 性能调优

# 生产者性能优化 class OptimizedProducer: def __init__(self): # 连接参数优化 self.connection = pika.BlockingConnection( pika.ConnectionParameters( host='localhost', heartbeat=600, blocked_connection_timeout=300, connection_attempts=3, retry_delay=5 ) ) # 启用发布确认 self.channel.confirm_delivery() # 开启事务（可选，会降低性能） # self.channel.tx_select() def batch_publish(self, messages): """批量发布消息""" for msg in messages: self.channel.basic_publish( exchange='batch.exchange', routing_key='batch.route', body=msg, properties=pika.BasicProperties( delivery_mode=2, content_type='application/json' ) ) # 批量确认 self.channel.publish_confirm_batch()

7.2 消费者优化

# 消费者性能优化 class OptimizedConsumer: def __init__(self): self.connection = pika.BlockingConnection( pika.ConnectionParameters(host='localhost') ) self.channel = self.connection.channel() # QoS配置 - 平衡延迟和吞吐量 self.channel.basic_qos(prefetch_count=100) def start_consuming(self): """高效消费""" self.channel.basic_consume( queue='optimized.queue', on_message_callback=self.efficient_callback, auto_ack=False ) self.channel.start_consuming() def efficient_callback(self, ch, method, properties, body): """批量处理消息""" # 批量处理而不是逐个处理 batch = [] while len(batch) < 100: # 获取消息 batch.append((method, properties, body)) # 非阻塞获取下一条 method, properties, body = self.channel.basic_get( queue='optimized.queue', auto_ack=False ) if body is None: break # 批量处理 self.process_batch(batch) # 批量确认 for method, _, _ in batch: ch.basic_ack(delivery_tag=method.delivery_tag)