商城系统通过Kafka消息队列，实现订单的处理和状态更新

本文介绍: createOr de r 方法中，我们首先将订单信息发送到“order”主题中，然后更新订单状态为“待支付”，并将订单状态更新消息发送到“order–status–update”主题中。在该方法中，我们获取订单状态更新消息，更新订单状态，并将订单信息更新到数据库中。当有新的下单请求到达时，我们会调用Or de rSe r v ic e的createOr de r 方法，将订单信息发送到Kaf k a中，并返回一个成功响应。在这里，我们指定了Kaf k a的地址和消费者组ID，并设置了生产者消息的序列化器和默认主题为“order”。

以下是一个简单的Spr in g Bo o t 应用程序示例，演示如何使用Kaf k a 实现订单的处理和状态更新。

首先，我们创建一个名为“order”的topic，在application.yaml 配置文件中添加Kaf k a的配置：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-group
      auto-offset-reset: earliest
    producer:
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    template:
      default-topic: order

在这里，我们指定了Kaf ka的地址和消费者组ID，并设置了生产者消息的序列化器和默认主题为“order”。

接下来，我们创建一个名为“Ord er”的Java类，来表示订单对象：

public class Order {
    private String orderId;
    private String userId;
    private String productId;
    private int quantity;
    private double price;
    private OrderStatus status;
    private String createTime;
    // getter、setter、toString
}

public enum OrderStatus {
    CREATED, PAID, CANCELLED, COMPLETED
}

public class OrderMessage {
    private Order order;
    private OrderStatus status;
    // getter、setter、toString
}

@Service
public class OrderService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void createOrder(Order order) {
        // 1、将订单信息发送到“order”主题中
        kafkaTemplate.send("order", order.toString());
        // 2、更新订单状态为“待支付”
        OrderMessage message = new OrderMessage();
        message.setOrder(order);
        message.setStatus(OrderStatus.CREATED);
        // 3、将订单状态更新消息发送到“order-status-update”主题中
        kafkaTemplate.send("order-status-update", message.toString());
    }

    @KafkaListener(topics = "order-status-update")
    public void handleOrderStatusUpdate(String message) {
        // 1、获取订单状态更新消息
        OrderMessage orderMessage = convertToOrderMessage(message);
        // 2、更新订单状态
        OrderStatus status = orderMessage.getStatus();
        Order order = orderMessage.getOrder();
        // 3、将订单信息更新到数据库中
        updateOrderStatus(order, status);
        // 4、如果订单状态为“已支付”，则执行后续操作
        if (status == OrderStatus.PAID) {
            // 生成物流信息
            generateLogisticsInfo(order);
            // 发送支付成功通知
            sendPaymentSuccessNotification(order);
        }
    }

    private OrderMessage convertToOrderMessage(String message) {
        // 解析订单状态更新消息
        ObjectMapper mapper = new ObjectMapper();
        try {
            return mapper.readValue(message, OrderMessage.class);
        } catch (Exception e) {
            throw new RuntimeException("Failed to convert message to OrderMessage: " + message, e);
        }
    }

    private void updateOrderStatus(Order order, OrderStatus status) {
        // 更新订单状态到数据库中
    }

    private void generateLogisticsInfo(Order order) {
        // 生成物流信息
    }

    private void sendPaymentSuccessNotification(Order order) {
        // 发送支付成功通知，可以通过发短信，邮箱通知
    }
}

@RestController
public class OrderController {
    @Autowired
    private OrderService orderService;

    @PostMapping("/order")
    public ResponseEntity<?> createOrder(@RequestBody Order order) {
        orderService.createOrder(order);
        return ResponseEntity.ok().build();
    }
}