RabbitMQ在订单服务中的使用【沙盘演示】

“世界上有一种人，和你在一起的时候，总是千万次嘱咐要多穿件衣服，要注意安全，你觉得很烦，却也觉得很温暖。” —— 钱钟书《围城》

概述

当处理实际的订单业务时，通常会涉及到多个步骤和不同的系统。以下是一个理论上的复杂订单处理流程，用于演示 RabbitMQ 如何在不同步骤之间协调和传递消息。在这个场景中，我们考虑了订单创建、库存管理、支付处理和通知服务等步骤。

订单处理流程：

1. 订单创建（Order Creation）：

   • 用户通过前端应用或网站创建新订单。
   • 订单系统将订单信息发布到 RabbitMQ 中，通知其他系统。

2. 库存检查（Inventory Check）：

   • 库存管理系统订阅订单创建消息，并检查商品库存。
   • 如果库存充足，库存系统发布库存确认消息；否则，发布库存不足消息。

3. 支付处理（Payment Processing）：

   • 支付服务订阅订单创建消息，获取订单信息。
   • 处理用户支付，更新订单支付状态。
   • 发布支付状态更新消息。

4. 订单确认（Order Confirmation）：

   • 订单系统订阅支付状态更新消息，获取订单信息。
   • 如果支付成功，发送订单确认邮件给用户。
   • 如果支付失败，发送支付失败通知。

5. 通知服务（Notification Service）：

   • 通知服务订阅订单确认消息，发送通知给相关团队或服务。

这里有进一步的思考，我们的库存检查到底是在创建队列前呢还是创建队列后呢？

在这里，我一直在犹豫，我认为是应该创建队列前进行库存检查，然后进行接下来的流程，不然的话，人家下单了，你告诉人家库存不足，这岂不是十分不恰！其实，确确实实应该是在订单创建完成，队列添加成功后进行判断库存，我刚刚只是自己的逻辑自洽！

其实仔细去考虑这个逻辑，用户之所以可以下单，肯定是在库存充裕的情况下才可以完成的，否则，前端也不会直接让下单，早就报出了库存不足的字样，根本完成不了下单这一个操作，所以，应该是先下单，后判断库存，而且生成提示，这个提示是给后台管理人员，而不是推送给用户的！

这样，我们就接着写剩下的逻辑——

代码实现

下面是一个简单的示例，使用 RabbitMQ 来模拟订单处理流程。

首先，确保你的项目中包含 RabbitMQ 的 Java 客户端库（通常使用 Maven 或 Gradle 进行依赖管理，当然我们经常使用SpringBoot，所以大家都知道怎么完成依赖引入）。

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class OrderSystem {
    private static final String ORDER_EXCHANGE = "order_exchange";
    private static final String ORDER_QUEUE = "order_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 定义交换机
            channel.exchangeDeclare(ORDER_EXCHANGE, BuiltinExchangeType.DIRECT);

            // 定义队列
            channel.queueDeclare(ORDER_QUEUE, false, false, false, null);

            // 绑定队列到交换机
            channel.queueBind(ORDER_QUEUE, ORDER_EXCHANGE, "order.created");

            // 消费订单消息
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println("[x] Received Order: " + message);

                // 模拟库存检查
                boolean isInventorySufficient = checkInventory(message);

                if (isInventorySufficient) {
                    System.out.println("[x] Inventory is sufficient. Processing payment...");
                    // 模拟支付处理
                    processPayment(message);

                    // 发布支付状态更新消息
                    channel.basicPublish(ORDER_EXCHANGE, "payment.updated", null, message.getBytes());
                } else {
                    System.out.println("[x] Insufficient inventory. Cancelling order...");
                    // 发布库存不足消息
                    channel.basicPublish(ORDER_EXCHANGE, "inventory.insufficient", null, message.getBytes());
                }
            };

            // 开始消费订单消息
            System.out.println("[*] Waiting for orders. To exit, press CTRL+C");
            channel.basicConsume(ORDER_QUEUE, true, deliverCallback, consumerTag -> {
            });
        }
    }

    private static boolean checkInventory(String orderMessage) {
        // 模拟库存检查逻辑，这里可以根据实际业务逻辑进行处理
        // 在这个简化的示例中，假设库存充足
        return true;
    }

    private static void processPayment(String orderMessage) {
        // 模拟支付处理逻辑，这里可以根据实际业务逻辑进行处理
        System.out.println("[x] Payment processed successfully.");
    }
}

上述代码是订单系统的部分，主要负责订单的创建、库存检查、支付处理等逻辑。接下来，你可以创建其他系统，如库存管理系统、支付服务、订单确认系统和通知服务，并在这些系统中使用相似的代码结构进行消息订阅和处理。

当然，我们知道RabbitMQ有很多模式，比如消费者订阅模式，路由模式啥的，这里我们用的是什么呢？

是的，你可以将这个模式看作是一种路由模式。在 RabbitMQ 中，Direct Exchange 是一种路由模式的 Exchange 类型。在路由模式中，消息的路由是根据消息的 Routing Key 进行的。

这段代码使用了 RabbitMQ 的 Direct Exchange 模式，具体体现在以下几个地方：

1. 交换机类型（Exchange Type）：

   channel.exchangeDeclare(ORDER_EXCHANGE, BuiltinExchangeType.DIRECT);
   

   在这里，使用了 BuiltinExchangeType.DIRECT，这表示创建了一个直连交换机。Direct Exchange 是一种简单的消息路由模式，它根据消息的 Routing Key 将消息直接发送到与之匹配的队列中。

2. 队列与交换机的绑定：

   channel.queueBind(ORDER_QUEUE, ORDER_EXCHANGE, "order.created");
   

   这里将队列 order_queue 绑定到直连交换机 order_exchange，并指定了一个 Routing Key 为 “order.created”。这表示只有使用相同 Routing Key 的消息才会被路由到该队列。

3. 消息发布：

   channel.basicPublish(ORDER_EXCHANGE, "payment.updated", null, message.getBytes());
   

   在订单系统接收到订单消息后，通过 channel.basicPublish 发布了一条消息到交换机 order_exchange，Routing Key 为 “payment.updated”。这样，消息会被路由到与该 Routing Key 绑定的队列。

总的来说，这个模式使用 Direct Exchange，通过指定消息的 Routing Key 来进行消息的路由。这种模式适用于简单的一对一或一对少数的消息通信场景。

在这个例子中，订单系统发布订单消息时指定的 Routing Key 是 “order.created”，而在消费者端通过 channel.queueBind(ORDER_QUEUE, ORDER_EXCHANGE, "order.created"); 绑定了队列和交换机，并指定了相同的 Routing Key。这样，只有携带 “order.created” Routing Key 的消息才会被路由到该队列进行消费。

因此，这个模式可以被视为一种基于 Routing Key 进行消息路由的路由模式。这种模式的特点是消息的目标队列是根据消息的路由键进行匹配的，适用于一对一或一对少数的消息通信场景。

为什么使用路由模式呢？

我们再赘述一下：

1. 明确的路由规则： Direct Exchange 模式使用 Routing Key 进行消息路由，这使得消息的目标队列可以非常明确和直观地指定。如果你有一组特定的消息需要路由到特定的队列，这种直连关系可以很容易地建立起来。

2. 简单明了： Direct Exchange 模式相对简单，易于理解和使用。当系统的消息路由需求较为简单，且不需要复杂的消息匹配规则时，Direct Exchange 提供了一种轻量级的解决方案。

3. 精确匹配： Direct Exchange 模式提供精确的匹配方式，只有指定 Routing Key 的消息会被路由到相应的队列。这对于确保消息只被发送到特定的接收者是有益的。

4. 适用于一对一或一对少数的场景： 如果你的消息通信场景是一对一或者一对少数的情况，Direct Exchange 是一种有效的选择。这样可以确保每条消息都能够准确地被路由到目标队列。

需要注意的是，如果你的消息路由逻辑更为复杂，需要使用一对多的模式，或者需要更灵活的匹配规则，可能需要考虑其他类型的 Exchange 模式，如Topic Exchange或Fanout Exchange。选择 Exchange 模式取决于你的具体业务需求和消息通信的复杂性。