系统讲解Apache Kafka消息管理与异常处理的最佳实践

// 生产者：发送带 Key 的消息，后续覆盖旧值
Properties props = new Properties();
props.put("bootstrap.servers", "kafka-server:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("ysx_mob_log", "key1", "new_value")); // 覆盖 key1 的旧消息
producer.close();

# 消费原 Topic，过滤错误数据后写入新 Topic
kafka-console-consumer.sh \
  --bootstrap-server kafka-server:9092 \
  --topic ysx_mob_log \
  --from-beginning \
  | grep -v "BAD_DATA" \
  | kafka-console-producer.sh \
    --bootstrap-server kafka-server:9092 \
    --topic ysx_mob_log_clean

// 使用 KafkaAdminClient 删除指定 Offset（Java 示例）
try (AdminClient admin = AdminClient.create(props)) {
    Map<TopicPartition, RecordsToDelete> records = new HashMap<>();
    records.put(new TopicPartition("ysx_mob_log", 0), RecordsToDelete.beforeOffset(100L));
    admin.deleteRecords(records).all().get(); // 删除 Partition 0 的 Offset <100 的消息
}

kafka-console-consumer.sh \
  --bootstrap-server kafka-server:9092 \
  --topic ysx_mob_log \
  --formatter "kafka.tools.DefaultMessageFormatter" \
  --property print.value=true \
  --property value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer \
  --skip-message-on-error  # 关键参数

public class SafeDeserializer implements Deserializer<String> {
    @Override
    public String deserialize(String topic, byte[] data) {
        try {
            return new String(data, StandardCharsets.UTF_8);
        } catch (Exception e) {
            System.err.println("Bad message: " + Arrays.toString(data));
            return null; // 返回 null 会被消费者跳过
        }
    }
}

// 消费者配置
props.put("value.deserializer", "com.example.SafeDeserializer");

// 生产者确保写入合法 JSON
ObjectMapper mapper = new ObjectMapper();
String json = mapper.writeValueAsString(new MyData(...)); // 使用 Jackson 序列化
producer.send(new ProducerRecord<>("ysx_mob_log", json));

kafka-console-consumer.sh \
  --bootstrap-server kafka-server:9092 \
  --topic ysx_mob_log \
  --property print.offset=true \
  --property print.value=false \
  --offset 0 --partition 0
# 输出示例: offset=100, value=[B@1a2b3c4d

# Python 消费者过滤二进制数据
from kafka import KafkaConsumer
consumer = KafkaConsumer(
    'ysx_mob_log',
    bootstrap_servers='kafka-server:9092',
    value_deserializer=lambda x: x.decode('utf-8') if x.startswith(b'{') else None
)
for msg in consumer:
    if msg.value: print(msg.value)  # 仅处理合法 JSON

// 生产者强制校验数据格式
public void sendToKafka(String data) {
    try {
        new ObjectMapper().readTree(data); // 校验是否为合法 JSON
        producer.send(new ProducerRecord<>("ysx_mob_log", data));
    } catch (Exception e) {
        log.error("Invalid JSON: {}", data);
    }
}