Kafka接收的数据是Base64编码字符串，即使生产者没有显式编码。

在生产者的代码中，需要将发送的消息转换为字节数组，而不是字符串。这样，在消费者消费数据时，就可以正确解码，而不是得到Base64编码的字符串。

以下是一个示例代码：

生产者：

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
import java.util.Base64;

public class MyProducer {
  public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("key.serializer", StringSerializer.class.getName());
    props.put("value.serializer", ByteArraySerializer.class.getName());
    Producer<String, byte[]> producer = new KafkaProducer<>(props);

    String message = "Hello World!";
    byte[] data = message.getBytes(StandardCharsets.UTF_8);
    producer.send(new ProducerRecord<>("my_topic", "my_key", data));

    producer.close();
  }
}

消费者：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Properties;
import java.util.Base64;

public class MyConsumer {
  public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("group.id", "my_group");
    props.put("key.deserializer", StringDeserializer.class.getName());
    props.put("value.deserializer", ByteArrayDeserializer.class.getName());
    Consumer<String, byte[]> consumer = new KafkaConsumer<>(props);
    consumer.subscribe(Collections.singletonList("my_topic"));

    while (true) {
      ConsumerRecords<String, byte[]> records = consumer.poll(100);
      for (ConsumerRecord<String, byte[]> record : records) {
        String key = record.key();
        byte[] value = record