Kafka架构与原理：高吞吐分布式日志系统

// RabbitMQ模型：Queue（单队列）
Queue: order.queue
    ├─ Message 1
    ├─ Message 2
    ├─ Message 3
    ...

限制：
- 单队列吞吐量受限（单机磁盘IO）
- 无法水平扩展

// Kafka模型：Topic + Partition（分区）
Topic: order
    ├─ Partition 0 → Broker 1
    │   ├─ Message 1
    │   ├─ Message 4
    │   ...
    ├─ Partition 1 → Broker 2
    │   ├─ Message 2
    │   ├─ Message 5
    │   ...
    └─ Partition 2 → Broker 3
        ├─ Message 3
        ├─ Message 6
        ...

优势：
- 并行处理：多个Partition可以并行读写
- 水平扩展：增加Partition可以提升吞吐量
- 负载均衡：Partition分布在不同Broker

/**
 * Kafka Producer示例：发送消息到分区
 */
public class KafkaProducerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {

            // 方式1：不指定Key（轮询分配到分区）
            for (int i = 0; i < 10; i++) {
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        "order-topic",  // Topic
                        "订单-" + i     // Value
                );

                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("发送失败: " + exception.getMessage());
                    } else {
                        System.out.println("发送成功 → Partition: " + metadata.partition()
                                + ", Offset: " + metadata.offset());
                    }
                });
            }

            // 方式2：指定Key（根据Key的Hash值分配到分区）
            for (int i = 0; i < 10; i++) {
                String key = "user-" + (i % 3);  // 3个用户
                ProducerRecord<String, String> record = new ProducerRecord<>(
                        "order-topic",  // Topic
                        key,            // Key（决定分区）
                        "订单-" + i     // Value
                );

                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("发送失败: " + exception.getMessage());
                    } else {
                        System.out.println("Key: " + key + " → Partition: " + metadata.partition()
                                + ", Offset: " + metadata.offset());
                    }
                });
            }

            // 方式3：自定义分区器
            // props.put("partitioner.class", "com.example.CustomPartitioner");
        }
    }
}

// Kafka将消息追加到日志文件末尾（顺序写）
public class SegmentAppend {

    private FileChannel fileChannel;
    private long offset;

    /**
     * 追加消息到日志文件（顺序写）
     */
    public void append(byte[] message) throws IOException {
        ByteBuffer buffer = ByteBuffer.wrap(message);

        // 顺序写到文件末尾
        fileChannel.write(buffer, offset);

        // 更新offset
        offset += message.length;
    }
}

/**
 * 零拷贝示例
 */
public class ZeroCopyExample {

    /**
     * 传统方式：4次拷贝
     */
    public void traditionalCopy(File file, SocketChannel socketChannel) throws IOException {
        try (FileInputStream fis = new FileInputStream(file)) {
            byte[] buffer = new byte[4096];
            int bytesRead;

            while ((bytesRead = fis.read(buffer)) != -1) {
                // 从磁盘读取到用户空间（2次拷贝）
                socketChannel.write(ByteBuffer.wrap(buffer, 0, bytesRead));
                // 从用户空间写入到网卡（2次拷贝）
            }
        }
    }

    /**
     * 零拷贝方式：2次拷贝（使用transferTo）
     */
    public void zeroCopy(File file, SocketChannel socketChannel) throws IOException {
        try (FileChannel fileChannel = new FileInputStream(file).getChannel()) {
            // 直接从磁盘发送到网卡（2次拷贝）
            fileChannel.transferTo(0, fileChannel.size(), socketChannel);
        }
    }
}

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");

// 批量发送配置
props.put("batch.size", 16384);  // 批次大小：16KB
props.put("linger.ms", 10);      // 等待时间：10ms
props.put("buffer.memory", 33554432);  // 缓冲区：32MB

// 原理：
// Producer会等待10ms或积累16KB消息后批量发送

// 压缩配置
props.put("compression.type", "lz4");  // 压缩算法：lz4（性能最好）

// 压缩率：
// - 原始大小：1MB
// - 压缩后：200KB（压缩率80%）
// - 网络传输：200KB（提升5倍）

// acks配置
props.put("acks", "all");  // 等待所有ISR副本确认

// acks取值：
// - acks = 0：不等待确认（最快，可能丢失）
// - acks = 1：等待Leader确认（中等，Leader故障可能丢失）
// - acks = all：等待所有ISR副本确认（最慢，不会丢失）

/**
 * Kafka Consumer示例
 */
public class KafkaConsumerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "order-consumer-group");  // Consumer Group
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 分区分配策略
        props.put("partition.assignment.strategy",
                "org.apache.kafka.clients.consumer.RangeAssignor");
        // - RangeAssignor：按范围分配（默认）
        // - RoundRobinAssignor：轮询分配
        // - StickyAssignor：粘性分配（尽量保持之前的分配）

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("order-topic"));

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

                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                            record.partition(), record.offset(), record.key(), record.value());

                    // 处理消息
                    processMessage(record.value());
                }

                // 手动提交offset
                consumer.commitSync();
            }
        }
    }

    private static void processMessage(String message) {
        // 业务逻辑
    }
}