当前位置：首页 > 学习笔记 > 正文内容

消息队列完全指南：分布式系统通信的核心实践

廖万里16小时前学习笔记0

"消息队列是分布式系统的核心组件，通过异步通信实现服务解耦、流量削峰和可靠传输。"

一、消息队列核心价值

消息队列是一种进程间通信方式，发送方将消息放入队列，接收方从队列中取出消息处理。这种异步通信模式为分布式系统带来了三大核心价值：异步处理、服务解耦、流量削峰。

1. 异步处理提升性能

传统同步调用模式下，用户请求需要等待所有操作完成才能返回响应。使用消息队列后，核心操作完成后即可返回，其他操作异步处理，响应时间从数秒降至毫秒级。

2. 服务解耦降低复杂度

消息队列引入后，服务只需发送消息到队列，不关心谁消费消息。新服务订阅消息即可接入系统，无需修改现有代码。

3. 流量削峰保证稳定性

消息队列作为缓冲区，将请求先存入队列，后端服务按处理能力消费消息，避免系统被瞬间高并发压垮。

二、主流消息队列对比

RabbitMQ

基于Erlang开发，实现了AMQP协议。可靠性高、功能丰富，支持多种消息模式，适合金融交易、订单处理场景。

Apache Kafka

分布式流处理平台，基于日志存储设计。吞吐量极高、支持消息持久化，适合日志收集、流处理场景。

Apache RocketMQ

阿里巴巴开源，专为金融互联网设计。支持事务消息、顺序消息，适合电商、金融场景。

三、RabbitMQ实战

交换机类型

// 直连交换机：精确匹配路由键
@Bean
public DirectExchange directExchange() {
    return new DirectExchange("exchange.direct");
}

// 主题交换机：支持通配符匹配
@Bean
public TopicExchange topicExchange() {
    return new TopicExchange("exchange.topic");
}

// 扇出交换机：广播到所有绑定队列
@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("exchange.fanout");
}

消息发送与消费

@Service
public class OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void createOrder(Order order) {
        orderRepository.save(order);
        rabbitTemplate.convertAndSend(
            "exchange.order",
            "order.created",
            new OrderCreatedEvent(order.getId())
        );
    }
}

@Component
public class InventoryConsumer {
    @RabbitListener(queues = "queue.inventory")
    public void handleOrderCreated(OrderCreatedEvent event) {
        inventoryService.deductStock(event.getOrderId());
    }
}

四、Kafka实战

生产者配置

@Configuration
public class KafkaProducerConfig {
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.ACKS_CONFIG, "all");
        config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        return new DefaultKafkaProducerFactory<>(config);
    }
}

消费者配置

@Service
public class OrderKafkaConsumer {
    @KafkaListener(topics = "order-events", groupId = "order-group")
    public void consume(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            OrderEvent event = JSON.parseObject(record.value(), OrderEvent.class);
            processOrder(event);
            ack.acknowledge();
        } catch (Exception e) {
            log.error("处理消息失败", e);
        }
    }
}

五、消息幂等性

消息可能被重复消费，需要在消费端实现幂等性。

@Service
public class IdempotentConsumer {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    
    public void processMessage(Message message) {
        String key = "msg:processed:" + message.getId();
        Boolean success = redisTemplate.opsForValue()
            .setIfAbsent(key, "1", Duration.ofHours(24));
        
        if (!success) {
            return;  // 已处理，跳过
        }
        
        doProcess(message);
    }
}

六、消息顺序性

Kafka保证同一分区内消息的顺序性。将需要顺序处理的消息发送到同一分区即可。

// 生产者：使用订单ID作为分区键
kafkaTemplate.send("order-events", order.getId(), orderEvent);

七、延迟消息

延迟消息用于定时任务场景，如订单超时取消。

// RabbitMQ使用TTL + 死信队列实现延迟
@Bean
public Queue delayQueue() {
    return QueueBuilder.durable("queue.delay")
        .ttl(30000)  // 30秒后过期
        .deadLetterExchange("exchange.dlx")
        .build();
}