单个Kafka消费者同时消费多无关联Key主题的聚合处理方案咨询

现有两个无键值关联的Kafka主题：

• topic1无Key，Payload示例：

{ "bookList": [{"bookId": "1"}, {"bookId": "2" } ],"magazineList": [{"magazineId": "1"}, {"magazineId": "2" } ]}

• topic2带有随机整数类型的Key，Payload示例：

{ "libraryId": "1", "cityId": "1" }

要求同时消费这两个主题（可使用Kafka Stream）并对Payload进行聚合处理，且主题属于同一消费者组。另外询问以下Java代码方案是否可行：

KafkaConsumer<String,String> Consumer = new KafkaConsumer<String,String>(properties);
       Consumer.subscribe("topic1");
       Consumer.subscribe("topic2");

        
        while (true) {
            ConsumerRecords<Integer,String> records=Consumer.poll(Duration.ofMillis(100));  
                 for(ConsumerRecord<String,String> record: records){  
            System.out.println(record);
            }
        }

一、你提供的基础消费者代码是否可行？ #

这段代码不可行，存在3个核心问题：

1. 订阅逻辑错误：连续调用subscribe会覆盖前一次的订阅，最终只会订阅topic2。要同时订阅多个主题，应该传入主题列表：subscribe(Arrays.asList("topic1", "topic2"))。
2. 泛型不匹配：声明的KafkaConsumer<String,String>与poll返回的ConsumerRecords<Integer,String>泛型类型冲突，编译会直接报错，正确写法应为ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));。
3. 无聚合能力：代码仅实现了记录打印，没有任何聚合逻辑，完全无法满足你的业务需求。

即便修正上述问题，基础消费者也只能实现简单多主题消费，手动维护聚合状态会非常繁琐，推荐使用Kafka Streams完成聚合。

二、用Kafka Streams实现多主题聚合的方案 #

Kafka Streams的聚合依赖Key关联，因此首先要确定聚合的关联维度（示例假设以cityId作为关联键，若topic1无此字段，需补充业务规则生成对应Key）。

步骤1：定义数据模型 #

创建POJO类用于序列化/反序列化Payload：

// Book.java
public class Book {
    private String bookId;
    // getter、setter、toString方法
}

// Magazine.java
public class Magazine {
    private String magazineId;
    // getter、setter、toString方法
}

// Topic1Payload.java
public class Topic1Payload {
    private List<Book> bookList;
    private List<Magazine> magazineList;
    // getter、setter、toString方法
}

// Topic2Payload.java
public class Topic2Payload {
    private String libraryId;
    private String cityId;
    // getter、setter、toString方法
}

步骤2：构建Kafka Streams拓扑 #

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.*;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;

public class MultiTopicAggregation {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "multi-topic-agg-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        JsonSerde<Topic1Payload> topic1Serde = new JsonSerde<>(Topic1Payload.class);
        JsonSerde<Topic2Payload> topic2Serde = new JsonSerde<>(Topic2Payload.class);

        // 处理topic1：生成统一Key（示例用默认城市Key，可根据业务调整）
        KStream<String, Topic1Payload> topic1Stream = builder.stream("topic1", Consumed.with(Serdes.String(), topic1Serde))
                .map((key, value) -> KeyValue.pair("default-city", value));

        // 处理topic2：提取cityId作为关联Key
        KStream<String, Topic2Payload> topic2Stream = builder.stream("topic2", Consumed.with(Serdes.Integer(), topic2Serde))
                .map((key, value) -> KeyValue.pair(value.getCityId(), value));

        // 合并两个流
        KStream<String, Object> mergedStream = topic1Stream.merge(topic2Stream);

        // 转换数据格式，适配聚合逻辑
        KStream<String, Integer> countStream = mergedStream.flatMapValues(value -> {
            if (value instanceof Topic1Payload) {
                Topic1Payload payload = (Topic1Payload) value;
                return Collections.singletonList(payload.getBookList().size() + payload.getMagazineList().size());
            } else if (value instanceof Topic2Payload) {
                return Collections.singletonList(1); // 每个图书馆记为1个单位
            }
            return Collections.emptyList();
        });

        // 按Key聚合，统计总数
        KTable<String, Long> aggregatedTable = countStream.groupByKey()
                .aggregate(
                        () -> 0L, // 初始值
                        (key, value, aggregate) -> aggregate + value, // 累加逻辑
                        Materialized.as("aggregation-store") // 状态存储名称
                );

        // 输出聚合结果（可替换为输出到Kafka主题）
        aggregatedTable.toStream().foreach((key, value) -> System.out.println("城市ID: " + key + ", 总关联数量: " + value));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();

        // 注册关闭钩子
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }

    // 自定义JSON序列化器
    static class JsonSerde<T> extends Serdes.WrapperSerde<T> {
        public JsonSerde(Class<T> type) {
            super(new org.apache.kafka.common.serialization.JsonSerializer<>(),
                  new org.apache.kafka.common.serialization.JsonDeserializer<>(type));
        }
    }
}

关键说明 #

1. 统一Key：Kafka Streams的聚合依赖相同Key将记录路由到同一分区，因此必须通过map操作将两个流的记录转换为同一维度的Key（如示例中的cityId）。
2. 状态管理：Kafka Streams自动维护聚合状态，无需手动处理，比基础消费者更可靠。
3. 序列化：自定义JsonSerde简化POJO的序列化/反序列化操作。

三、关于“不同主题需使用相同Key”的说明 #

这句话的核心逻辑是：若要对多主题记录做关联或聚合，必须让需要关联的记录拥有相同Key，这样Kafka才能保证同Key的记录被路由到同一分区处理，确保聚合结果的正确性。如果原主题无相同Key，需通过业务规则生成统一Key后再进行聚合。

内容的提问来源于stack exchange，提问作者yigi