我正在评估apachekafka流的事件源，看看它对于复杂场景的可行性。与关系数据库一样，我遇到过一些情况，原子性/事务性是必不可少的：
提供两种服务的购物应用程序：
orderservice：有一个kafka流存储和orders（ordersstore）
产品服务：有一个Kafka流商店（productstockstore）和产品及其库存。
流量：
orderservice发布ordercreated事件（带有productid、orderid、userid信息）
productservice获取ordercreated事件并查询其kafkastreams存储（productstockstore）以检查产品是否有库存。如果有库存，它会发布orderupdated事件（还包括productid、orderid、userid信息）
关键是这个事件将由productservice kafka stream监听，它将处理它以减少库存，到目前为止还不错。
但是，想象一下：
客户1下订单，订单1（产品的库存为1）
客户2同时为同一产品下另一个订单order2（库存仍然是1）
productservice处理order1并发送消息orderupdated以减少库存。此消息放在order2->ordercreated的主题之后
productservice处理order2 ordercreated并发送消息orderupdated以再次减少库存。这是不正确的，因为它会导致不一致（现在库存应该是0）。
显而易见的问题是，当我们处理第一个orderupdated事件时，我们的物化视图（存储）应该直接更新。但是，更新kafka流存储的唯一方法（我知道）是发布另一个事件（orderupdated），由kafka流处理。这样我们就不能事务性地执行此更新。
我希望能有一些想法来处理这样的情况。
更新：我将尝试澄清问题中有问题的部分：
productservice有一个kafka streams商店，productstock有这个库存 (productId=1, quantity=1) orderservice在orders主题上发布两个orderplaced事件：
Event1 (key=product1, productId=product1, quantity=1, eventType="OrderPlaced") Event2 (key=product1, productId=product1, quantity=1, eventType="OrderPlaced") productservice在orders主题上有一个使用者。为了简单起见，让我们假设一个分区来保证消息按顺序使用。此使用者执行以下逻辑：

if("OrderPlaced".equals(event.get("eventType"))){

    Order order = new Order();
    order.setId((String)event.get("orderId"));
    order.setProductId((Integer)(event.get("productId")));
    order.setUid(event.get("uid").toString());

    // QUERY PRODUCTSTOCK TO CHECK AVAILABILITY
    Integer productStock = getProductStock(order.getProductId());

    if(productStock > 0) {
        Map<String, Object> event = new HashMap<>();
        event.put("name", "ProductReserved");
        event.put("orderId", order.getId());
        event.put("productId", order.getProductId());

        // WRITES A PRODUCT RESERVED EVENT TO orders topic
        orderProcessor.output().send(MessageBuilder.withPayload(event).build(), 500);
    }else{
        //XXX CANCEL ORDER
    }
}

productservice还有一个kafka streams处理器，负责更新库存：

KStream<Integer, JsonNode> stream = kStreamBuilder.stream(integerSerde, jsonSerde, "orders");
stream.xxx().yyy(() -> {...}, "ProductsStock");

event1将首先被处理，因为仍然有1个可用产品，所以它将生成productreserved事件。
现在，轮到event2了。如果在productservice kafka streams处理器处理由event1生成的productreseved事件之前，它被productservice使用者使用，则使用者仍会看到product1的productstore库存为1，为event2生成productreserved事件，然后在系统中产生不一致。