Kafka 面试宝典

// 创建只有 1 个分区的 Topic
kafka-topics.sh --create --topic orders \
  --partitions 1 --replication-factor 3

// 同一用户的订单发送到同一分区
producer.send(new ProducerRecord<>(
    "orders",           // topic
    userId,             // key（决定分区）
    orderData           // value
));

// 分区算法：hash(key) % partition_count
// userId=1001 → Partition 0
// userId=1002 → Partition 0
// userId=2001 → Partition 1

public class OrderPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                        Object value, byte[] valueBytes, Cluster cluster) {
        // 根据订单类型分区
        String orderType = extractOrderType(value);
        return orderType.hashCode() % cluster.partitionCountForTopic(topic);
    }
}

// 1. acks=all（或-1）：等待所有 ISR 副本确认
props.put("acks", "all");

// 2. 重试次数
props.put("retries", 3);

// 3. 幂等性（避免重复）
props.put("enable.idempotence", true);

// 手动提交 offset（先处理，后提交）
props.put("enable.auto.commit", false);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 处理业务逻辑
        processRecord(record);
    }
    // 处理成功后再提交
    consumer.commitSync();
}

props.put("enable.idempotence", true);

// 原理：
// Producer 为每条消息分配 Sequence Number
// Broker 检测到重复的 Sequence Number 会自动去重

消息1: PID=100, Seq=0 → Broker 接收
消息1: PID=100, Seq=0 → Broker 检测重复，丢弃
消息2: PID=100, Seq=1 → Broker 接收

// 配置事务 ID
props.put("transactional.id", "order-tx-001");

Producer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();

    // 发送消息
    producer.send(new ProducerRecord<>("orders", order1));
    producer.send(new ProducerRecord<>("payments", payment1));

    // 提交事务
    producer.commitTransaction();
} catch (Exception e) {
    // 回滚事务
    producer.abortTransaction();
}

StreamsConfig config = new StreamsConfig(props);
config.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
           StreamsConfig.EXACTLY_ONCE_V2);

// 自动管理事务和 offset 提交

// 默认 10s，可适当增加
props.put("session.timeout.ms", 30000);

// 心跳间隔（应 < session.timeout.ms / 3）
props.put("heartbeat.interval.ms", 3000);

// 避免消费者处理时间过长，导致心跳超时
props.put("max.poll.records", 100);  // 默认 500

// 增加两次 poll 的最大间隔
props.put("max.poll.interval.ms", 300000);  // 5分钟

// 只重新分配必要的分区，避免全量停止
props.put("partition.assignment.strategy",
          "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

// Consumer 重启后不触发 Rebalance
props.put("group.instance.id", "consumer-1");

// Producer 批量发送配置
props.put("batch.size", 16384);           // 16KB
props.put("linger.ms", 10);               // 等待 10ms 凑批次
props.put("compression.type", "lz4");     // 压缩

Properties props = new Properties();

// 1. 批量发送
props.put("batch.size", 32768);          // 32KB（增大批次）
props.put("linger.ms", 20);              // 等待 20ms

// 2. 压缩
props.put("compression.type", "lz4");    // lz4 压缩比和性能平衡

// 3. 缓冲区
props.put("buffer.memory", 67108864);    // 64MB（增大缓冲）

// 4. 并发请求
props.put("max.in.flight.requests.per.connection", 5);

// 5. acks 配置（根据场景选择）
props.put("acks", "1");                  // 只等 Leader 确认（高吞吐）
// props.put("acks", "all");             // 等所有 ISR 确认（高可靠）

// 每次拉取更多数据
props.put("fetch.min.bytes", 1048576);       // 1MB
props.put("fetch.max.wait.ms", 500);         // 最多等 500ms
props.put("max.poll.records", 500);          // 每次拉取 500 条

// 增大接收缓冲区
props.put("receive.buffer.bytes", 65536);    // 64KB

// 方案一：一个 Consumer，多线程处理（推荐）
ExecutorService executor = Executors.newFixedThreadPool(10);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    for (ConsumerRecord<String, String> record : records) {
        executor.submit(() -> {
            processRecord(record);
        });
    }

    consumer.commitSync();  // 全部处理完再提交
}

// 方案二：多个 Consumer 实例（需注意线程安全）

props.put("enable.auto.commit", false);

// 批量提交 offset（减少与 Broker 交互）
consumer.commitSync();

// RocketMQ 支持 18 个延迟级别
message.setDelayTimeLevel(3);  // 延迟 10s

// 降低处理复杂度，快速消费
// 将耗时操作异步化或暂时跳过
if (isBacklog()) {
    asyncProcess(record);  // 异步处理
} else {
    syncProcess(record);   // 正常同步处理
}

// 生产者：保证消息不丢
props.put("acks", "all");
props.put("enable.idempotence", true);

// 消费者：手动提交，先处理后提交
props.put("enable.auto.commit", false);
props.put("isolation.level", "read_committed");  // 只读已提交的事务消息