07-性能优化与调优

// 批量处理配置
type BatchProcessor struct {
    batchSize          int           // 批次大小
    lingerMs           int           // 延迟时间
    compressionType    string        // 压缩类型
    bufferMemory       int64         // 缓冲区内存
    maxRequestSize     int32         // 最大请求大小
    retries            int           // 重试次数
    maxInFlightRequests int          // 最大飞行请求数
}

// 批量发送实现
func (bp *BatchProcessor) sendBatch(records []Record) error {
    // 1. 检查批次大小
    if len(records) < bp.batchSize && time.Since(bp.lastSendTime) < time.Duration(bp.lingerMs)*time.Millisecond {
        return nil // 等待更多记录
    }
    
    // 2. 压缩批次
    compressedBatch, err := bp.compressBatch(records)
    if err != nil {
        return err
    }
    
    // 3. 发送批次
    return bp.sendCompressedBatch(compressedBatch)
}

// 压缩批次
func (bp *BatchProcessor) compressBatch(records []Record) ([]byte, error) {
    switch bp.compressionType {
    case "gzip":
        return bp.compressGzip(records)
    case "snappy":
        return bp.compressSnappy(records)
    case "lz4":
        return bp.compressLZ4(records)
    case "zstd":
        return bp.compressZstd(records)
    default:
        return bp.compressNone(records)
    }
}

// Gzip压缩
func (bp *BatchProcessor) compressGzip(records []Record) ([]byte, error) {
    var buf bytes.Buffer
    writer := gzip.NewWriter(&buf)
    
    for _, record := range records {
        data, err := json.Marshal(record)
        if err != nil {
            return nil, err
        }
        if _, err := writer.Write(data); err != nil {
            return nil, err
        }
    }
    
    if err := writer.Close(); err != nil {
        return nil, err
    }
    
    return buf.Bytes(), nil
}

// Snappy压缩
func (bp *BatchProcessor) compressSnappy(records []Record) ([]byte, error) {
    var buf bytes.Buffer
    writer := snappy.NewWriter(&buf)
    
    for _, record := range records {
        data, err := json.Marshal(record)
        if err != nil {
            return nil, err
        }
        if _, err := writer.Write(data); err != nil {
            return nil, err
        }
    }
    
    if err := writer.Close(); err != nil {
        return nil, err
    }
    
    return buf.Bytes(), nil
}

// 异步发送器
type AsyncSender struct {
    buffer      chan *ProducerRecord
    workers     int
    batchSize   int
    lingerMs    int
    compression string
    mu          sync.Mutex
    batches     map[string]*Batch // Topic-Partition -> Batch
}

// 异步发送
func (as *AsyncSender) sendAsync(record *ProducerRecord) error {
    // 1. 添加到缓冲区
    select {
    case as.buffer <- record:
        return nil
    default:
        return fmt.Errorf("缓冲区已满")
    }
}

// 批量处理工作器
func (as *AsyncSender) batchWorker() {
    ticker := time.NewTicker(time.Duration(as.lingerMs) * time.Millisecond)
    defer ticker.Stop()
    
    for {
        select {
        case record := <-as.buffer:
            as.addToBatch(record)
        case <-ticker.C:
            as.flushBatches()
        }
    }
}

// 添加到批次
func (as *AsyncSender) addToBatch(record *ProducerRecord) {
    as.mu.Lock()
    defer as.mu.Unlock()
    
    key := fmt.Sprintf("%s-%d", record.Topic, record.Partition)
    batch := as.batches[key]
    
    if batch == nil {
        batch = &Batch{
            Topic:     record.Topic,
            Partition: record.Partition,
            Records:   make([]*ProducerRecord, 0),
        }
        as.batches[key] = batch
    }
    
    batch.Records = append(batch.Records, record)
    
    // 检查是否达到批次大小
    if len(batch.Records) >= as.batchSize {
        as.flushBatch(batch)
        delete(as.batches, key)
    }
}

// 刷新批次
func (as *AsyncSender) flushBatches() {
    as.mu.Lock()
    defer as.mu.Unlock()
    
    for key, batch := range as.batches {
        if len(batch.Records) > 0 {
            as.flushBatch(batch)
            delete(as.batches, key)
        }
    }
}

// 刷新单个批次
func (as *AsyncSender) flushBatch(batch *Batch) {
    // 1. 压缩批次
    compressedData, err := as.compressBatch(batch.Records)
    if err != nil {
        log.Printf("压缩批次失败: %v", err)
        return
    }
    
    // 2. 发送批次
    go func() {
        if err := as.sendBatch(batch.Topic, batch.Partition, compressedData); err != nil {
            log.Printf("发送批次失败: %v", err)
        }
    }()
}

// 分区选择器
type PartitionSelector struct {
    partitioner string
    hashFunc    func([]byte) int32
}

// 选择分区
func (ps *PartitionSelector) selectPartition(topic string, key []byte, partitionCount int32) int32 {
    switch ps.partitioner {
    case "hash":
        return ps.hashPartition(key, partitionCount)
    case "roundrobin":
        return ps.roundRobinPartition(partitionCount)
    case "random":
        return ps.randomPartition(partitionCount)
    default:
        return ps.hashPartition(key, partitionCount)
    }
}

// 哈希分区
func (ps *PartitionSelector) hashPartition(key []byte, partitionCount int32) int32 {
    if key == nil {
        return 0
    }
    
    hash := ps.hashFunc(key)
    return hash % partitionCount
}

// 轮询分区
func (ps *PartitionSelector) roundRobinPartition(partitionCount int32) int32 {
    ps.mu.Lock()
    defer ps.mu.Unlock()
    
    partition := ps.currentPartition
    ps.currentPartition = (ps.currentPartition + 1) % partitionCount
    
    return partition
}

// 随机分区
func (ps *PartitionSelector) randomPartition(partitionCount int32) int32 {
    return int32(rand.Intn(int(partitionCount)))
}

// 网络线程配置
type NetworkThreadConfig struct {
    numNetworkThreads    int
    numIOThreads         int
    socketSendBuffer     int
    socketReceiveBuffer  int
    maxConnectionsPerIP  int
    connectionsMaxIdleMs int
}

// 网络线程管理器
type NetworkThreadManager struct {
    config        *NetworkThreadConfig
    acceptor      *Acceptor
    processors    []*Processor
    requestQueue  chan *Request
    responseQueue chan *Response
}

// 启动网络线程
func (ntm *NetworkThreadManager) start() error {
    // 1. 启动接受器
    if err := ntm.acceptor.start(); err != nil {
        return err
    }
    
    // 2. 启动处理器
    for i := 0; i < ntm.config.numNetworkThreads; i++ {
        processor := &Processor{
            id:           i,
            requestQueue: ntm.requestQueue,
            responseQueue: ntm.responseQueue,
        }
        ntm.processors = append(ntm.processors, processor)
        go processor.start()
    }
    
    return nil
}

// 处理器
type Processor struct {
    id           int
    requestQueue chan *Request
    responseQueue chan *Response
    handler      *RequestHandler
}

// 启动处理器
func (p *Processor) start() {
    for {
        select {
        case request := <-p.requestQueue:
            p.handleRequest(request)
        }
    }
}

// 处理请求
func (p *Processor) handleRequest(request *Request) {
    // 1. 解析请求
    parsedRequest, err := p.parseRequest(request)
    if err != nil {
        p.sendErrorResponse(request, err)
        return
    }
    
    // 2. 处理请求
    response, err := p.handler.handle(parsedRequest)
    if err != nil {
        p.sendErrorResponse(request, err)
        return
    }
    
    // 3. 发送响应
    p.sendResponse(request, response)
}

// IO线程配置
type IOThreadConfig struct {
    numIOThreads        int
    queuedMaxRequests   int
    requestTimeoutMs    int
    maxRequestSize      int32
    replicaFetchMaxBytes int32
}

// IO线程管理器
type IOThreadManager struct {
    config        *IOThreadConfig
    ioThreads     []*IOThread
    requestQueue  chan *IORequest
    responseQueue chan *IOResponse
}

// IO线程
type IOThread struct {
    id           int
    requestQueue chan *IORequest
    responseQueue chan *IOResponse
    logManager   *LogManager
}

// 启动IO线程
func (iot *IOThread) start() {
    for {
        select {
        case request := <-iot.requestQueue:
            iot.handleIORequest(request)
        }
    }
}

// 处理IO请求
func (iot *IOThread) handleIORequest(request *IORequest) {
    switch request.Type {
    case ProduceRequest:
        iot.handleProduceRequest(request)
    case FetchRequest:
        iot.handleFetchRequest(request)
    case CommitOffsetRequest:
        iot.handleCommitOffsetRequest(request)
    default:
        iot.sendErrorResponse(request, fmt.Errorf("未知请求类型"))
    }
}

// 处理生产请求
func (iot *IOThread) handleProduceRequest(request *IORequest) {
    // 1. 验证请求
    if err := iot.validateProduceRequest(request); err != nil {
        iot.sendErrorResponse(request, err)
        return
    }
    
    // 2. 追加到日志
    offset, err := iot.logManager.append(request.Topic, request.Partition, request.Records)
    if err != nil {
        iot.sendErrorResponse(request, err)
        return
    }
    
    // 3. 发送响应
    response := &IOResponse{
        Type:       ProduceResponse,
        Topic:      request.Topic,
        Partition:  request.Partition,
        Offset:     offset,
        Error:      nil,
    }
    iot.sendResponse(request, response)
}

// 内存管理器
type MemoryManager struct {
    heapSize        int64
    pageCacheSize   int64
    bufferPool      *BufferPool
    gcThreshold     float64
    gcInterval      time.Duration
}

// 缓冲区池
type BufferPool struct {
    pool    sync.Pool
    size    int
    maxSize int
}

// 获取缓冲区
func (bp *BufferPool) get() []byte {
    if buf := bp.pool.Get(); buf != nil {
        return buf.([]byte)
    }
    return make([]byte, bp.size)
}

// 归还缓冲区
func (bp *BufferPool) put(buf []byte) {
    if len(buf) <= bp.maxSize {
        bp.pool.Put(buf)
    }
}

// 内存监控
func (mm *MemoryManager) monitorMemory() {
    ticker := time.NewTicker(mm.gcInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            mm.checkMemoryUsage()
        }
    }
}

// 检查内存使用
func (mm *MemoryManager) checkMemoryUsage() {
    var m runtime.MemStats
    runtime.ReadMemStats(&m)
    
    // 计算内存使用率
    usage := float64(m.Alloc) / float64(mm.heapSize)
    
    if usage > mm.gcThreshold {
        // 触发GC
        runtime.GC()
        log.Printf("触发GC，内存使用率: %.2f%%", usage*100)
    }
}

// 拉取策略配置
type FetchStrategy struct {
    fetchMinBytes     int32
    fetchMaxBytes     int32
    fetchMaxWaitMs    int32
    maxPartitionFetchBytes int32
    sessionTimeoutMs  int32
    heartbeatIntervalMs int32
}

// 拉取管理器
type FetchManager struct {
    strategy      *FetchStrategy
    consumerGroup string
    coordinator   *GroupCoordinator
    partitions    map[string][]int32
    offsets       map[string]map[int32]int64
}

// 拉取消息
func (fm *FetchManager) fetchMessages() ([]*ConsumerRecord, error) {
    // 1. 构建拉取请求
    request := fm.buildFetchRequest()
    
    // 2. 发送拉取请求
    response, err := fm.sendFetchRequest(request)
    if err != nil {
        return nil, err
    }
    
    // 3. 处理响应
    records, err := fm.processFetchResponse(response)
    if err != nil {
        return nil, err
    }
    
    // 4. 更新偏移量
    fm.updateOffsets(records)
    
    return records, nil
}

// 构建拉取请求
func (fm *FetchManager) buildFetchRequest() *FetchRequest {
    request := &FetchRequest{
        ConsumerGroup: fm.consumerGroup,
        MaxWaitMs:     fm.strategy.fetchMaxWaitMs,
        MinBytes:      fm.strategy.fetchMinBytes,
        MaxBytes:      fm.strategy.fetchMaxBytes,
        Topics:        make(map[string][]int32),
    }
    
    for topic, partitions := range fm.partitions {
        request.Topics[topic] = partitions
    }
    
    return request
}

// 处理拉取响应
func (fm *FetchManager) processFetchResponse(response *FetchResponse) ([]*ConsumerRecord, error) {
    var records []*ConsumerRecord
    
    for topic, partitionData := range response.Topics {
        for partition, data := range partitionData {
            if data.Error != nil {
                return nil, data.Error
            }
            
            // 解析记录
            partitionRecords, err := fm.parseRecords(topic, partition, data.Records)
            if err != nil {
                return nil, err
            }
            
            records = append(records, partitionRecords...)
        }
    }
    
    return records, nil
}

// 并行消费管理器
type ParallelConsumerManager struct {
    workers        int
    partitionQueue chan *PartitionTask
    resultQueue    chan *ConsumeResult
    consumer       *Consumer
}

// 分区任务
type PartitionTask struct {
    Topic     string
    Partition int32
    Offset    int64
    MaxBytes  int32
}

// 消费结果
type ConsumeResult struct {
    Topic     string
    Partition int32
    Records   []*ConsumerRecord
    Error     error
}

// 启动并行消费
func (pcm *ParallelConsumerManager) start() error {
    // 1. 启动工作器
    for i := 0; i < pcm.workers; i++ {
        go pcm.worker(i)
    }
    
    // 2. 启动任务分发器
    go pcm.taskDispatcher()
    
    // 3. 启动结果收集器
    go pcm.resultCollector()
    
    return nil
}

// 工作器
func (pcm *ParallelConsumerManager) worker(id int) {
    for task := range pcm.partitionQueue {
        // 1. 拉取消息
        records, err := pcm.consumer.fetchFromPartition(task.Topic, task.Partition, task.Offset, task.MaxBytes)
        
        // 2. 处理消息
        if err == nil {
            for _, record := range records {
                if err := pcm.processRecord(record); err != nil {
                    log.Printf("处理消息失败: %v", err)
                }
            }
        }
        
        // 3. 发送结果
        result := &ConsumeResult{
            Topic:     task.Topic,
            Partition: task.Partition,
            Records:   records,
            Error:     err,
        }
        pcm.resultQueue <- result
    }
}

// 任务分发器
func (pcm *ParallelConsumerManager) taskDispatcher() {
    for {
        // 1. 获取分区列表
        partitions := pcm.getPartitions()
        
        // 2. 创建任务
        for topic, partitionList := range partitions {
            for _, partition := range partitionList {
                task := &PartitionTask{
                    Topic:     topic,
                    Partition: partition,
                    Offset:    pcm.getOffset(topic, partition),
                    MaxBytes:  1024 * 1024, // 1MB
                }
                pcm.partitionQueue <- task
            }
        }
        
        // 3. 等待一段时间
        time.Sleep(100 * time.Millisecond)
    }
}

// Offset管理器
type OffsetManager struct {
    offsets        map[string]map[int32]int64
    commitInterval time.Duration
    autoCommit     bool
    mu             sync.RWMutex
}

// 自动提交Offset
func (om *OffsetManager) autoCommitOffsets() {
    ticker := time.NewTicker(om.commitInterval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            if om.autoCommit {
                om.commitOffsets()
            }
        }
    }
}

// 提交Offset
func (om *OffsetManager) commitOffsets() error {
    om.mu.RLock()
    offsets := make(map[string]map[int32]int64)
    for topic, partitions := range om.offsets {
        offsets[topic] = make(map[int32]int64)
        for partition, offset := range partitions {
            offsets[topic][partition] = offset
        }
    }
    om.mu.RUnlock()
    
    // 发送提交请求
    return om.sendCommitRequest(offsets)
}

// 更新Offset
func (om *OffsetManager) updateOffset(topic string, partition int32, offset int64) {
    om.mu.Lock()
    defer om.mu.Unlock()
    
    if om.offsets[topic] == nil {
        om.offsets[topic] = make(map[int32]int64)
    }
    
    om.offsets[topic][partition] = offset
}

// 网络优化配置
type NetworkOptimizer struct {
    tcpNoDelay        bool
    tcpKeepAlive      bool
    tcpKeepAliveIdle  int
    tcpKeepAliveIntvl int
    tcpKeepAliveProbes int
    socketBufferSize  int
    maxConnections    int
}

// 优化TCP连接
func (no *NetworkOptimizer) optimizeTCPConnection(conn net.Conn) error {
    if tcpConn, ok := conn.(*net.TCPConn); ok {
        // 设置TCP_NODELAY
        if err := tcpConn.SetNoDelay(no.tcpNoDelay); err != nil {
            return err
        }
        
        // 设置TCP_KEEPALIVE
        if err := tcpConn.SetKeepAlive(no.tcpKeepAlive); err != nil {
            return err
        }
        
        // 设置KeepAlive参数
        if err := tcpConn.SetKeepAlivePeriod(time.Duration(no.tcpKeepAliveIdle) * time.Second); err != nil {
            return err
        }
    }
    
    return nil
}

// 优化Socket缓冲区
func (no *NetworkOptimizer) optimizeSocketBuffer(conn net.Conn) error {
    if tcpConn, ok := conn.(*net.TCPConn); ok {
        // 设置发送缓冲区
        if err := tcpConn.SetWriteBuffer(no.socketBufferSize); err != nil {
            return err
        }
        
        // 设置接收缓冲区
        if err := tcpConn.SetReadBuffer(no.socketBufferSize); err != nil {
            return err
        }
    }
    
    return nil
}

// IO优化器
type IOOptimizer struct {
    useZeroCopy      bool
    useDirectIO      bool
    useAsyncIO       bool
    ioThreads        int
    queueDepth       int
}

// 零拷贝读取
func (ioo *IOOptimizer) zeroCopyRead(file *os.File, offset int64, size int64) ([]byte, error) {
    if ioo.useZeroCopy {
        // 使用sendfile系统调用
        return ioo.sendfileRead(file, offset, size)
    } else {
        // 使用普通读取
        return ioo.normalRead(file, offset, size)
    }
}

// sendfile读取
func (ioo *IOOptimizer) sendfileRead(file *os.File, offset int64, size int64) ([]byte, error) {
    // 创建内存映射
    data, err := syscall.Mmap(int(file.Fd()), offset, int(size), syscall.PROT_READ, syscall.MAP_SHARED)
    if err != nil {
        return nil, err
    }
    
    return data, nil
}

// 异步IO
func (ioo *IOOptimizer) asyncIO() {
    if ioo.useAsyncIO {
        // 使用异步IO
        go ioo.asyncIOWorker()
    }
}

// 异步IO工作器
func (ioo *IOOptimizer) asyncIOWorker() {
    for {
        // 处理异步IO请求
        select {
        case request := <-ioo.asyncIOQueue:
            ioo.handleAsyncIORequest(request)
        }
    }
}

// 性能指标
type PerformanceMetrics struct {
    // Producer指标
    ProducerThroughput    float64 // 生产吞吐量 (records/sec)
    ProducerLatency       float64 // 生产延迟 (ms)
    ProducerBatchSize     float64 // 平均批次大小
    
    // Broker指标
    BrokerCPUUsage        float64 // CPU使用率
    BrokerMemoryUsage     float64 // 内存使用率
    BrokerDiskUsage       float64 // 磁盘使用率
    BrokerNetworkIO       float64 // 网络IO
    
    // Consumer指标
    ConsumerLag           int64   // 消费延迟
    ConsumerThroughput    float64 // 消费吞吐量
    ConsumerFetchLatency  float64 // 拉取延迟
    
    // 集群指标
    UnderReplicatedPartitions int32  // 未充分复制分区数
    ISRShrinksPerSec      float64 // ISR收缩频率
    LeaderElectionsPerSec float64 // Leader选举频率
}

// 指标收集器
type MetricsCollector struct {
    metrics    *PerformanceMetrics
    collectors map[string]MetricCollector
    interval   time.Duration
}

// 收集指标
func (mc *MetricsCollector) collectMetrics() {
    ticker := time.NewTicker(mc.interval)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            mc.collectAllMetrics()
        }
    }
}

// 收集所有指标
func (mc *MetricsCollector) collectAllMetrics() {
    // 1. 收集Producer指标
    mc.collectProducerMetrics()
    
    // 2. 收集Broker指标
    mc.collectBrokerMetrics()
    
    // 3. 收集Consumer指标
    mc.collectConsumerMetrics()
    
    // 4. 收集集群指标
    mc.collectClusterMetrics()
}

// 收集Producer指标
func (mc *MetricsCollector) collectProducerMetrics() {
    // 计算吞吐量
    mc.metrics.ProducerThroughput = mc.calculateProducerThroughput()
    
    // 计算延迟
    mc.metrics.ProducerLatency = mc.calculateProducerLatency()
    
    // 计算批次大小
    mc.metrics.ProducerBatchSize = mc.calculateProducerBatchSize()
}

// 收集Broker指标
func (mc *MetricsCollector) collectBrokerMetrics() {
    // 获取系统指标
    mc.metrics.BrokerCPUUsage = mc.getCPUUsage()
    mc.metrics.BrokerMemoryUsage = mc.getMemoryUsage()
    mc.metrics.BrokerDiskUsage = mc.getDiskUsage()
    mc.metrics.BrokerNetworkIO = mc.getNetworkIO()
}

// 瓶颈分析器
type BottleneckAnalyzer struct {
    metrics *PerformanceMetrics
    thresholds map[string]float64
}

// 分析瓶颈
func (ba *BottleneckAnalyzer) analyzeBottlenecks() []string {
    var bottlenecks []string
    
    // 1. 检查CPU瓶颈
    if ba.metrics.BrokerCPUUsage > ba.thresholds["cpu"] {
        bottlenecks = append(bottlenecks, "CPU使用率过高")
    }
    
    // 2. 检查内存瓶颈
    if ba.metrics.BrokerMemoryUsage > ba.thresholds["memory"] {
        bottlenecks = append(bottlenecks, "内存使用率过高")
    }
    
    // 3. 检查磁盘瓶颈
    if ba.metrics.BrokerDiskUsage > ba.thresholds["disk"] {
        bottlenecks = append(bottlenecks, "磁盘使用率过高")
    }
    
    // 4. 检查网络瓶颈
    if ba.metrics.BrokerNetworkIO > ba.thresholds["network"] {
        bottlenecks = append(bottlenecks, "网络IO过高")
    }
    
    // 5. 检查消费延迟
    if ba.metrics.ConsumerLag > int64(ba.thresholds["consumer_lag"]) {
        bottlenecks = append(bottlenecks, "消费延迟过高")
    }
    
    return bottlenecks
}

// 提供优化建议
func (ba *BottleneckAnalyzer) provideOptimizationSuggestions(bottlenecks []string) []string {
    var suggestions []string
    
    for _, bottleneck := range bottlenecks {
        switch bottleneck {
        case "CPU使用率过高":
            suggestions = append(suggestions, "增加CPU核心数或优化代码")
        case "内存使用率过高":
            suggestions = append(suggestions, "增加内存或优化内存使用")
        case "磁盘使用率过高":
            suggestions = append(suggestions, "增加磁盘容量或优化存储")
        case "网络IO过高":
            suggestions = append(suggestions, "增加网络带宽或优化网络配置")
        case "消费延迟过高":
            suggestions = append(suggestions, "增加消费者数量或优化消费逻辑")
        }
    }
    
    return suggestions
}