Kafka流：如何在同一个主题中处理不同对象类型的消息？

Kafka流可以使用Serde（序列化/反序列化）来处理不同的对象类型。Serde是一种用于在流中序列化和反序列化对象的方式。它可以将对象序列化为字节数组，反之亦然。使用不同的Serde，可以在同一个主题中处理不同的对象类型，并在处理它们的过程中保持灵活性。

以下是一个示例，该示例读取包含两种不同对象类型的数据的Kafka主题，并在处理它们时使用不同的Serde：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.Consumed;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.Serialized;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.util.Properties;

public class DifferentObjectsInSameTopicExample {

  public static void main(final String[] args) {
    final Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "different-objects-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.ByteArray().getClass());

    final StreamsBuilder builder = new StreamsBuilder();

    final KStream<String, byte[]> input = builder.stream("my-input-topic");

    final KStream<String, String> stringStream = 
      input.filter((key, value) -> isStringValue(value))
           .mapValues(value -> new String(value))
           .through("string-output-topic", Produced.with(Serdes.String(), Serdes.String()));

    final KStream<String, Integer> integerStream = 
      input.filter((key, value) -> isIntegerValue(value))
           .mapValues(value -> ByteBuffer.wrap(value).getInt())
           .through("