使用Spring KafkaStreams组合来自两个主题的数据

Question

我正在研究一个解决方案，我必须将Kafka两个主题t1和t2的数据结合起来
t1将包含消息的前半部分，t2将包含消息的后半部分
例如，如果完整消息是“a1b1c1d1”和“a2b2c2d2”，那么
t1将有“a1b1”和“a2b2”
t2将有“c1d1”和“c2d2”
并且我必须对它们执行并集以生成“a1b1c1d1”和“a2b2c2d2”
，因为消息不会按顺序存储在KStream store中，并在它们变为有空<现在的问题是，这是一个好的解决方案吗？如果是，有一个例子。

@Component
public class MyViews {

    @Autowired
    public void buildProductView(StreamsBuilder sb) {
        sb.stream("products", Consumed.with(Serdes.Integer(), Serdes.serdeFrom(Product.class)))
                .map((k, v) -> new KeyValue<>(v.getId(), v))
                .toTable(Materialized.<Integer, Product, KeyValueStore<Bytes, byte[]>>as("products-view")
                        .withKeySerde(Serdes.Integer())
                        .withValueSerde(Serdes.serdeFrom(Product.class)));
    }

    @Autowired
    public void buildPriceView(StreamsBuilder sb) {
        sb.stream("prices", Consumed.with(Serdes.Integer(), Serdes.serdeFrom(Price.class)))
                .map((k, v) -> new KeyValue<>(v.getId(), v))
                .toTable(Materialized.<Integer, Price, KeyValueStore<Bytes, byte[]>>as("prices-view")
                        .withKeySerde(Serdes.Integer())
                        .withValueSerde(Serdes.serdeFrom(Price.class)));
    }

    @Data
    private class Product {
        Integer id;
        String item;
    }

    @Data
    private class Price {
        Integer id;
        Integer price;
    }

    @Data
    private class Order{
        Integer id;
        String item;
        Integer price;
    }
}

这就是我想要实现的
每当html" target="_blank">消息发布到“订单”主题时，我想检查“价格视图”中是否存在相应的ID，如果存在，那么我将组合它们生成“订单”对象。
类似地，当消息发布到“价格”主题时，我想检查“产品视图”

Answer 1

@Component
public class MyViews {

    @Autowired
    public void buildProductView(StreamsBuilder sb) {
        final Serde<Product> productSerde = Serdes.serdeFrom(new JsonSerializer<>(), new ProductDeserializer());
        final Serde<Price> valueSerde = Serdes.serdeFrom(new JsonSerializer<>(), new PriceDeserializer());

        final KTable<Integer, Product> products = sb.stream("products", Consumed.with(Serdes.Integer(), productSerde)).selectKey((k, v) -> v.getId())
                .toTable(Materialized.<Integer, Product, KeyValueStore<Bytes, byte[]>>as("products-view")
                        .withKeySerde(Serdes.Integer())
                        .withValueSerde(productSerde)
                        .withCachingDisabled());

        final KTable<Integer, Price> prices = sb.stream("prices", Consumed.with(Serdes.Integer(), valueSerde))
                .map((k, v) -> new KeyValue<>(v.getId(), v))
                .toTable(Materialized.<Integer, Price, KeyValueStore<Bytes, byte[]>>as("prices-view")
                        .withKeySerde(Serdes.Integer())
                        .withValueSerde(valueSerde)
                        .withCachingDisabled());

        final KTable<Integer, Order> join = products.join(prices, Product::getId, (l, r) ->
                new Order(l.getId(), l.getItem(), r.getPrice())
        );

        join.toStream().foreach((x, y) -> System.out.println(y));

    }
    

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Product {
        Integer id;
        String item;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Price {
        Integer id;
        Integer price;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    @ToString
    public class Order {
        Integer id;
        String item;
        Integer price;
    }
}