// Kafka日志段核心源码分析
public class LogSegment {
    private final File logFile;
    private final File indexFile;
    private final long baseOffset;

    // 消息追加操作
    public void append(long offset, ByteBuffer message) {
        // 计算物理位置
        int physicalPosition = translateOffset(offset);
        // 写入日志文件
        logFile.write(message, physicalPosition);
        // 更新索引
        updateIndex(offset, physicalPosition);
    }

    // 偏移量转换
    private int translateOffset(long offset) {
        return (int) (offset - baseOffset) * RECORD_SIZE;
    }
}

public class HeartbeatManager {
    private final ScheduledExecutorService scheduler;
    private final Map<String, Long> lastHeartbeatTimes;
    private final long heartbeatTimeout;

    public void startHeartbeat(String nodeId) {
        scheduler.scheduleAtFixedRate(() -> {
            long currentTime = System.currentTimeMillis();
            Long lastTime = lastHeartbeatTimes.get(nodeId);

            if (lastTime != null && 
                currentTime - lastTime > heartbeatTimeout) {
                // 触发故障转移
                handleNodeFailure(nodeId);
            }
        }, 0, heartbeatInterval, TimeUnit.MILLISECONDS);
    }
}

public class MessageIndex {
    private final NavigableMap<Long, IndexEntry> sparseIndex;
    private final int indexInterval;

    // 稀疏索引构建
    public void addIndexEntry(long offset, long physicalPosition) {
        if (offset % indexInterval == 0) {
            sparseIndex.put(offset, new IndexEntry(physicalPosition));
        }
    }

    // 范围查询优化
    public List<IndexEntry> rangeQuery(long startOffset, long endOffset) {
        // 使用稀疏索引快速定位
        Long floorKey = sparseIndex.floorKey(startOffset);
        // 顺序扫描目标段
        return sequentialScan(floorKey, endOffset);
    }
}

public class ReplicaManager {
    // 副本同步核心方法
    public void appendRecords(
        long baseOffset,
        MemoryRecords records,
        boolean isFromClient,
        AppendCallback callback) {

        // 领导者副本写入本地日志
        LogAppendInfo appendInfo = log.append(records);

        // 异步复制到追随者
        if (isFromClient) {
            replicaFetcherManager.scheduleReplication(
                appendInfo, records, callback);
        }
    }

    // 副本选举算法
    public void electNewLeader(Partition partition) {
        // 从ISR中选择新领导者
        List<Replica> inSyncReplicas = partition.inSyncReplicas();
        Replica newLeader = selectEligibleLeader(inSyncReplicas);

        // 更新元数据
        updateLeaderAndIsr(partition, newLeader);

        // 通知所有副本
        notifyAllReplicas(partition, newLeader);
    }
}

% Erlang源码分析
handle_call({publish, Message}, _From, State) ->
    % 主节点处理消息发布
    {ok, LocalResult} = rabbit_amqqueue:deliver([Message], false),

    % 同步到镜像节点
    case State#state.gm of
        undefined ->
            {reply, LocalResult, State};
        GM ->
            % 使用GM协议广播
            gm:broadcast(GM, {publish, Message}),
            {reply, LocalResult, State}
    end.

public class ZeroCopyTransfer {
    // 使用FileChannel transferTo实现零拷贝
    public long transferTo(FileChannel fileChannel, 
                          long position, long count,
                          WritableByteChannel target) {
        return fileChannel.transferTo(position, count, target);
    }

    // 内存映射文件读取
    public MappedByteBuffer mapFile(File file, long position, long size) {
        try (FileChannel channel = FileChannel.open(file.toPath())) {
            return channel.map(FileChannel.MapMode.READ_ONLY, 
                             position, size);
        }
    }
}

# Docker Compose配置
version: '3.8'
services:
  rabbitmq:
    image: rabbitmq:3.9-management
    environment:

      - RABBITMQ_DEFAULT_USER=admin
      - RABBITMQ_DEFAULT_PASS=secret
    ports:

      - "5672:5672"
      - "15672:15672"
    volumes:

      - rabbitmq_data:/var/lib/rabbitmq

volumes:
  rabbitmq_data:

# broker核心配置
broker.id=1
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://hostname:9092

# 副本与可用性配置
default.replication.factor=3
min.insync.replicas=2
unclean.leader.election.enable=false

# 性能优化配置
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 日志存储配置
log.dirs=/kafka/logs
log.segment.bytes=1073741824
log.retention.hours=168
log.cleanup.policy=delete

Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
producerProps.put("acks", "all");
producerProps.put("retries", 10);
producerProps.put("batch.size", 16384);
producerProps.put("linger.ms", 10);
producerProps.put("buffer.memory", 33554432);
producerProps.put("compression.type", "lz4");

Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
consumerProps.put("group.id", "my-consumer-group");
consumerProps.put("enable.auto.commit", "false");
consumerProps.put("auto.offset.reset", "earliest");
consumerProps.put("max.poll.records", 500);

public class CustomMetricsCollector {
    private final MeterRegistry meterRegistry;

    @EventListener
    public void handleMessageEvent(MessageEvent event) {
        // 记录消息处理指标
        meterRegistry.counter("messages.processed", 
            "topic", event.getTopic(),
            "status", event.getStatus()
        ).increment();

        // 记录处理延迟
        Timer.builder("message.processing.time")
            .tag("topic", event.getTopic())
            .register(meterRegistry)
            .record(event.getProcessingTime());
    }
}

# Kafka调试命令示例
# 查看主题详情
kafka-topics.sh --describe --topic my-topic

# 查看消费者组偏移量
kafka-consumer-groups.sh --describe --group my-group

# 生产性能测试
kafka-producer-perf-test.sh --topic test --num-records 1000000 \
    --record-size 1000 --throughput 100000 --producer-props \
    bootstrap.servers=localhost:9092

# 消费性能测试  
kafka-consumer-perf-test.sh --topic test --messages 1000000 \
    --broker-list localhost:9092