Kafka的JsonDeserializer无法正常工作于java.util.Map。

要解决Kafka的JsonDeserializer无法正常工作于java.util.Map的问题，可以使用自定义的JsonDeserializer来处理。以下是一个示例代码：

首先，创建一个自定义的JsonDeserializer类，继承自Kafka自带的JsonDeserializer类，并指定泛型为Map类型：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;

public class MapJsonDeserializer implements Deserializer<Map<String, Object>> {

    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public Map<String, Object> deserialize(String topic, byte[] data) {
        try {
            return objectMapper.readValue(data, Map.class);
        } catch (Exception e) {
            throw new RuntimeException("Error deserializing JSON", e);
        }
    }

    @Override
    public void close() {
        // 关闭资源的操作（如果有的话）
    }
}

然后，在消费者代码中使用自定义的JsonDeserializer类来处理Map类型的数据：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Map;
import java.util.Properties;

public class ConsumerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MapJsonDeserializer.class.getName());

        KafkaConsumer<String, Map<String, Object>> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        while (true) {
            ConsumerRecords<String, Map<String, Object>> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, Map<String, Object>> record : records) {
                // 处理消息记录
                Map<String, Object> data = record.value();
                System.out.println("Received message: " + data);
            }
        }
    }
}

在上述代码中，我们将值的反序列化器设置为自定义的MapJsonDeserializer类，它将字节数组反序列化为Map<String, Object>对象。然后，在消费者代码中，我们可以直接使用Map<String, Object>来处理接收到的消息。