Istio Mixer遥测数据处理深度剖析：属性转换与策略执行的核心机制

graph TB
    subgraph "Mixer 组件架构"
        subgraph "API层 (API Layer)"
            mixer_api[Mixer API<br/>gRPC接口]
            check_api[Check API<br/>策略检查]
            report_api[Report API<br/>遥测上报]
        end
        
        subgraph "核心处理层 (Core Processing)"
            attr_processor[Attribute Processor<br/>属性处理器]
            template_engine[Template Engine<br/>模板引擎]
            adapter_manager[Adapter Manager<br/>适配器管理器]
            config_manager[Config Manager<br/>配置管理器]
        end
        
        subgraph "属性系统 (Attribute System)"
            attr_bag[Attribute Bag<br/>属性包]
            attr_vocab[Attribute Vocabulary<br/>属性词汇表]
            expr_evaluator[Expression Evaluator<br/>表达式求值器]
        end
        
        subgraph "适配器层 (Adapter Layer)"
            prometheus_adapter[Prometheus<br/>指标适配器]
            stackdriver_adapter[Stackdriver<br/>日志适配器]
            policy_adapter[Policy<br/>策略适配器]
            custom_adapter[Custom<br/>自定义适配器]
        end
        
        subgraph "后端系统 (Backend Systems)"
            prometheus[Prometheus<br/>指标存储]
            logging[Logging<br/>日志系统]
            monitoring[Monitoring<br/>监控系统]
            policy_engine[Policy Engine<br/>策略引擎]
        end
    end
    
    %% 数据流
    mixer_api --> attr_processor
    check_api --> template_engine
    report_api --> adapter_manager
    
    attr_processor --> attr_bag
    attr_bag --> attr_vocab
    attr_vocab --> expr_evaluator
    
    template_engine --> adapter_manager
    adapter_manager --> prometheus_adapter
    adapter_manager --> stackdriver_adapter
    adapter_manager --> policy_adapter
    adapter_manager --> custom_adapter
    
    prometheus_adapter -.->|指标上报| prometheus
    stackdriver_adapter -.->|日志上报| logging
    policy_adapter -.->|策略执行| policy_engine
    
    style attr_processor fill:#e1f5fe
    style template_engine fill:#f3e5f5
    style adapter_manager fill:#e8f5e8
    style attr_bag fill:#fff3e0

// mixer/pkg/attribute/mutableBag.go - 核心属性处理逻辑
type MutableBag struct {
    // 父属性包 - 支持属性继承
    parent Bag
    
    // 字符串属性映射
    strings map[string]string
    
    // 整数属性映射
    int64s map[string]int64
    
    // 浮点数属性映射
    float64s map[string]float64
    
    // 布尔属性映射
    bools map[string]bool
    
    // 时间属性映射
    times map[string]time.Time
    
    // 字节数组属性映射
    bytes map[string][]uint8
    
    // 字符串映射属性
    stringMaps map[string]map[string]string
    
    // 互斥锁保护并发访问
    lock sync.RWMutex
}

// UpdateBagFromProto - 从Protobuf消息更新属性包
// 这是Mixer属性处理的核心函数，负责将压缩的属性转换为标准词汇
func (mb *MutableBag) UpdateBagFromProto(attrs *mixerpb.CompressedAttributes, globalWordList []string) error {
    // 1. 获取消息词汇表
    messageWordList := attrs.Words
    
    // 2. 构建完整的词汇查找表
    // globalWordList是全局共享的词汇表，messageWordList是消息特定的词汇表
    totalWords := len(globalWordList) + len(messageWordList)
    wordList := make([]string, totalWords)
    copy(wordList[:len(globalWordList)], globalWordList)
    copy(wordList[len(globalWordList):], messageWordList)
    
    var e error
    
    // 3. 处理字符串属性
    for k, v := range attrs.Strings {
        // 查找属性名和值对应的词汇
        name, nameErr := lookup(k, globalWordList, messageWordList)
        value, valueErr := lookup(v, globalWordList, messageWordList)
        
        if nameErr != nil || valueErr != nil {
            e = multierror.Append(e, fmt.Errorf("string attribute lookup failed: name=%v, value=%v", nameErr, valueErr))
            continue
        }
        
        // 插入到属性包中
        if err := mb.insertProtoAttr(name, value, attributeTypeString); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 4. 处理整数属性
    for k, v := range attrs.Int64s {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("int64 attribute name lookup failed: %v", nameErr))
            continue
        }
        
        if err := mb.insertProtoAttr(name, v, attributeTypeInt64); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 5. 处理浮点数属性
    for k, v := range attrs.Float64s {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("float64 attribute name lookup failed: %v", nameErr))
            continue
        }
        
        if err := mb.insertProtoAttr(name, v, attributeTypeFloat64); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 6. 处理布尔属性
    for k, v := range attrs.Bools {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("bool attribute name lookup failed: %v", nameErr))
            continue
        }
        
        if err := mb.insertProtoAttr(name, v, attributeTypeBool); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 7. 处理时间戳属性
    for k, v := range attrs.Timestamps {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("timestamp attribute name lookup failed: %v", nameErr))
            continue
        }
        
        // 转换Protobuf时间戳为Go时间
        t := time.Unix(v.Seconds, int64(v.Nanos))
        if err := mb.insertProtoAttr(name, t, attributeTypeTime); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 8. 处理字节数组属性
    for k, v := range attrs.Bytes {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("bytes attribute name lookup failed: %v", nameErr))
            continue
        }
        
        if err := mb.insertProtoAttr(name, v, attributeTypeBytes); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    // 9. 处理字符串映射属性
    for k, v := range attrs.StringMaps {
        name, nameErr := lookup(k, globalWordList, messageWordList)
        if nameErr != nil {
            e = multierror.Append(e, fmt.Errorf("stringMap attribute name lookup failed: %v", nameErr))
            continue
        }
        
        // 转换值映射
        valueMap := make(map[string]string)
        for vk, vv := range v.Entries {
            valueKey, vkErr := lookup(vk, globalWordList, messageWordList)
            valueVal, vvErr := lookup(vv, globalWordList, messageWordList)
            
            if vkErr != nil || vvErr != nil {
                e = multierror.Append(e, fmt.Errorf("stringMap value lookup failed: key=%v, val=%v", vkErr, vvErr))
                continue
            }
            
            valueMap[valueKey] = valueVal
        }
        
        if err := mb.insertProtoAttr(name, valueMap, attributeTypeStringMap); err != nil {
            e = multierror.Append(e, err)
        }
    }
    
    return e
}

// 词汇查找核心函数
func lookup(index int32, globalList, messageList []string) (string, error) {
    if index < 0 {
        return "", fmt.Errorf("negative index %d", index)
    }
    
    // 先查找全局词汇表
    if index < int32(len(globalList)) {
        return globalList[index], nil
    }
    
    // 再查找消息词汇表
    messageIndex := index - int32(len(globalList))
    if messageIndex < int32(len(messageList)) {
        return messageList[messageIndex], nil
    }
    
    return "", fmt.Errorf("index %d out of range (global: %d, message: %d)", 
                        index, len(globalList), len(messageList))
}

// mixer/template/template.gen.go - 属性预处理实现
func Preprocess(ctx context.Context, attrs attribute.Bag, mapper expr.AttributeDescriptorFinder) (*Instance, error) {
    // 1. 创建实例模板
    instance := &Instance{}
    
    // 2. 在Kubernetes环境下追加特定属性
    if isKubernetesEnvironment(ctx) {
        // 追加目标服务的标签信息
        if destinationLabels, found := attrs.Get("destination.labels"); found {
            if labelsMap, ok := destinationLabels.(map[string]string); ok {
                instance.DestinationLabels = labelsMap
            }
        }
        
        // 追加目标服务的命名空间
        if destinationNamespace, found := attrs.Get("destination.namespace"); found {
            if ns, ok := destinationNamespace.(string); ok {
                instance.DestinationNamespace = ns
            }
        }
        
        // 追加源服务的标签信息
        if sourceLabels, found := attrs.Get("source.labels"); found {
            if labelsMap, ok := sourceLabels.(map[string]string); ok {
                instance.SourceLabels = labelsMap
            }
        }
        
        // 追加源服务的命名空间
        if sourceNamespace, found := attrs.Get("source.namespace"); found {
            if ns, ok := sourceNamespace.(string); ok {
                instance.SourceNamespace = ns
            }
        }
        
        // 追加请求ID（用于链路追踪）
        if requestID, found := attrs.Get("request.id"); found {
            if id, ok := requestID.(string); ok {
                instance.RequestID = id
            }
        }
        
        // 追加用户身份信息
        if userPrincipal, found := attrs.Get("request.auth.principal"); found {
            if principal, ok := userPrincipal.(string); ok {
                instance.UserPrincipal = principal
            }
        }
    }
    
    // 3. 计算派生属性
    instance.RequestTotalSize = calculateRequestSize(attrs)
    instance.ResponseTotalSize = calculateResponseSize(attrs)
    instance.RequestDuration = calculateDuration(attrs)
    
    // 4. 应用表达式求值
    if err := evaluateExpressions(instance, attrs, mapper); err != nil {
        return nil, fmt.Errorf("expression evaluation failed: %v", err)
    }
    
    return instance, nil
}

// 计算请求大小
func calculateRequestSize(attrs attribute.Bag) int64 {
    var totalSize int64
    
    // 请求头大小
    if requestHeaders, found := attrs.Get("request.headers"); found {
        if headers, ok := requestHeaders.(map[string]string); ok {
            for k, v := range headers {
                totalSize += int64(len(k) + len(v) + 4) // key: value\r\n
            }
        }
    }
    
    // 请求体大小
    if requestSize, found := attrs.Get("request.size"); found {
        if size, ok := requestSize.(int64); ok {
            totalSize += size
        }
    }
    
    return totalSize
}

// 表达式求值器
func evaluateExpressions(instance *Instance, attrs attribute.Bag, mapper expr.AttributeDescriptorFinder) error {
    // 创建求值上下文
    evalCtx := &expr.EvalContext{
        Attrs:  attrs,
        Mapper: mapper,
    }
    
    // 求值预定义表达式
    expressions := []struct {
        field *string
        expr  string
    }{
        {&instance.ServiceName, `destination.service.name | "unknown"`},
        {&instance.ServiceVersion, `destination.service.version | "unknown"`},
        {&instance.RequestProtocol, `request.protocol | "unknown"`},
        {&instance.ResponseCode, `response.code | 0`},
    }
    
    for _, e := range expressions {
        result, err := expr.Eval(e.expr, evalCtx)
        if err != nil {
            return fmt.Errorf("failed to evaluate expression %s: %v", e.expr, err)
        }
        
        if strResult, ok := result.(string); ok {
            *e.field = strResult
        }
    }
    
    return nil
}

// mixer/pkg/api/grpcServer.go - Check API实现
func (s *grpcServer) Check(ctx context.Context, req *mixerpb.CheckRequest) (*mixerpb.CheckResponse, error) {
    // 1. 记录请求指标
    s.monitoring.ChecksTotal.Increment()
    
    // 2. 属性包处理
    protoBag := attribute.GetMutableBag(nil)
    defer protoBag.Done()
    
    // 3. 从请求中提取和转换属性
    if err := protoBag.UpdateBagFromProto(&req.Attributes, s.globalWordList); err != nil {
        s.monitoring.CheckErrors.Increment()
        return nil, status.Errorf(codes.InvalidArgument, "failed to update attribute bag: %v", err)
    }
    
    // 4. 执行属性预处理
    if err := s.preprocessor.Preprocess(ctx, protoBag); err != nil {
        s.monitoring.CheckErrors.Increment()
        return nil, status.Errorf(codes.Internal, "preprocessing failed: %v", err)
    }
    
    // 5. 策略检查调度
    checkResult, err := s.dispatcher.Check(ctx, protoBag)
    if err != nil {
        s.monitoring.CheckErrors.Increment()
        return nil, status.Errorf(codes.Internal, "check dispatch failed: %v", err)
    }
    
    // 6. 构建响应
    response := &mixerpb.CheckResponse{
        Status: checkResult.Status,
        ValidDuration: checkResult.ValidDuration,
        ValidUseCount: checkResult.ValidUseCount,
    }
    
    // 7. 记录成功指标
    s.monitoring.ChecksSuccessful.Increment()
    
    return response, nil
}

// 策略检查调度器
type Dispatcher struct {
    // 模板到适配器的映射
    routes map[string][]AdapterRoute
    
    // 表达式求值器缓存
    evaluatorCache *lru.Cache
    
    // 并发控制
    checkSemaphore chan struct{}
}

// 执行策略检查
func (d *Dispatcher) Check(ctx context.Context, bag attribute.Bag) (*CheckResult, error) {
    // 1. 并发控制
    select {
    case d.checkSemaphore <- struct{}{}:
        defer func() { <-d.checkSemaphore }()
    case <-ctx.Done():
        return nil, ctx.Err()
    }
    
    // 2. 遍历所有路由规则
    var results []*CheckResult
    
    for template, routes := range d.routes {
        for _, route := range routes {
            // 3. 检查条件匹配
            if !d.evaluateCondition(route.Condition, bag) {
                continue
            }
            
            // 4. 创建模板实例
            instance, err := d.createTemplateInstance(template, bag)
            if err != nil {
                return nil, fmt.Errorf("failed to create template instance: %v", err)
            }
            
            // 5. 调用适配器执行检查
            result, err := route.Adapter.Check(ctx, instance)
            if err != nil {
                return nil, fmt.Errorf("adapter check failed: %v", err)
            }
            
            results = append(results, result)
            
            // 6. 如果有拒绝结果，立即返回
            if result.Status.Code != int32(codes.OK) {
                return result, nil
            }
        }
    }
    
    // 7. 合并所有检查结果
    return d.mergeCheckResults(results), nil
}

// mixer/pkg/api/grpcServer.go - Report API实现
func (s *grpcServer) Report(ctx context.Context, req *mixerpb.ReportRequest) (*mixerpb.ReportResponse, error) {
    // 1. 记录遥测请求指标
    s.monitoring.ReportsTotal.Increment()
    
    // 2. 批量处理遥测数据
    var reportInstances []*ReportInstance
    
    for i, attrs := range req.Attributes {
        // 3. 创建属性包
        protoBag := attribute.GetMutableBag(nil)
        
        // 4. 从Protobuf更新属性
        if err := protoBag.UpdateBagFromProto(attrs, s.globalWordList); err != nil {
            log.Errorf("Failed to update bag from proto for report %d: %v", i, err)
            continue
        }
        
        // 5. 属性预处理 - 追加Kubernetes环境特定属性
        if err := s.preprocessor.Preprocess(ctx, protoBag); err != nil {
            log.Errorf("Preprocessing failed for report %d: %v", i, err)
            protoBag.Done()
            continue
        }
        
        // 6. 创建报告实例
        instance := &ReportInstance{
            Bag:       protoBag,
            Timestamp: time.Now(),
            RequestID: extractRequestID(protoBag),
        }
        
        reportInstances = append(reportInstances, instance)
    }
    
    // 7. 异步分发遥测数据
    go func() {
        if err := s.dispatcher.Report(ctx, reportInstances); err != nil {
            log.Errorf("Report dispatch failed: %v", err)
            s.monitoring.ReportErrors.Increment()
        } else {
            s.monitoring.ReportsSuccessful.Increment()
        }
        
        // 8. 清理资源
        for _, instance := range reportInstances {
            instance.Bag.Done()
        }
    }()
    
    return &mixerpb.ReportResponse{}, nil
}

// 遥测数据预处理器
type Preprocessor struct {
    // Kubernetes API客户端
    kubeClient kubernetes.Interface
    
    // 服务缓存
    serviceCache *ServiceCache
    
    // Pod标签缓存
    podLabelCache *PodLabelCache
}

// Kubernetes环境下的属性预处理
func (p *Preprocessor) Preprocess(ctx context.Context, bag attribute.MutableBag) error {
    // 1. 获取目标服务信息
    if destinationUID, found := bag.Get("destination.uid"); found {
        if uid, ok := destinationUID.(string); ok {
            service, err := p.serviceCache.GetServiceByUID(uid)
            if err == nil && service != nil {
                // 追加目标服务标签
                bag.Set("destination.labels", service.Labels)
                bag.Set("destination.namespace", service.Namespace)
                bag.Set("destination.service.name", service.Name)
                bag.Set("destination.service.version", service.Labels["version"])
            }
        }
    }
    
    // 2. 获取源服务信息
    if sourceUID, found := bag.Get("source.uid"); found {
        if uid, ok := sourceUID.(string); ok {
            pod, err := p.podLabelCache.GetPodByUID(uid)
            if err == nil && pod != nil {
                // 追加源服务标签
                bag.Set("source.labels", pod.Labels)
                bag.Set("source.namespace", pod.Namespace)
                bag.Set("source.workload.name", pod.Labels["app"])
                bag.Set("source.workload.namespace", pod.Namespace)
            }
        }
    }
    
    // 3. 计算派生属性
    if err := p.computeDerivedAttributes(bag); err != nil {
        return fmt.Errorf("failed to compute derived attributes: %v", err)
    }
    
    return nil
}

// mixer/pkg/adapter/adapter.go - 适配器接口定义
type Handler interface {
    // Close关闭适配器并清理资源
    Close() error
}

// 策略检查适配器接口
type CheckHandler interface {
    Handler
    
    // HandleCheck执行策略检查
    HandleCheck(ctx context.Context, instance interface{}) (CheckResult, error)
}

// 遥测报告适配器接口
type ReportHandler interface {
    Handler
    
    // HandleReport处理遥测数据
    HandleReport(ctx context.Context, instances []interface{}) error
}

// 配额管理适配器接口
type QuotaHandler interface {
    Handler
    
    // HandleQuota处理配额请求
    HandleQuota(ctx context.Context, instance interface{}, args QuotaArgs) (QuotaResult, error)
}

// 适配器构建器接口
type Builder interface {
    // Build根据配置创建适配器实例
    Build(ctx context.Context, env Env) (Handler, error)
    
    // SetAdapterConfig设置适配器特定配置
    SetAdapterConfig(cfg adapter.Config)
    
    // Validate验证配置有效性
    Validate() *config.Error
}

// mixer/adapter/prometheus/prometheus.go - Prometheus适配器实现
type builder struct {
    adapterConfig *config.Params    // 适配器配置参数
    metricTypes   []MetricType      // 支持的指标类型
}

func (b *builder) Build(ctx context.Context, env adapter.Env) (adapter.Handler, error) {
    // 1. 创建Prometheus注册器
    registry := prometheus.NewRegistry()
    
    // 2. 创建适配器实例
    adapter := &prometheusAdapter{
        env:      env,
        registry: registry,
        metrics:  make(map[string]prometheus.Collector),
        config:   b.adapterConfig,
    }
    
    // 3. 注册预定义指标
    for _, metricType := range b.metricTypes {
        if err := adapter.registerMetric(metricType); err != nil {
            return nil, fmt.Errorf("failed to register metric %s: %v", metricType.Name, err)
        }
    }
    
    // 4. 启动HTTP服务器暴露指标
    if err := adapter.startMetricsServer(); err != nil {
        return nil, fmt.Errorf("failed to start metrics server: %v", err)
    }
    
    return adapter, nil
}

type prometheusAdapter struct {
    env      adapter.Env
    registry *prometheus.Registry
    metrics  map[string]prometheus.Collector
    config   *config.Params
    server   *http.Server
}

// 处理遥测报告
func (p *prometheusAdapter) HandleReport(ctx context.Context, instances []interface{}) error {
    for _, instance := range instances {
        metricInstance, ok := instance.(*metric.Instance)
        if !ok {
            continue
        }
        
        // 1. 查找对应的Prometheus指标
        collector, exists := p.metrics[metricInstance.Name]
        if !exists {
            log.Warnf("Unknown metric: %s", metricInstance.Name)
            continue
        }
        
        // 2. 构建标签
        labels := make(prometheus.Labels)
        for k, v := range metricInstance.Dimensions {
            if strVal, ok := v.(string); ok {
                labels[k] = strVal
            }
        }
        
        // 3. 根据指标类型更新值
        switch c := collector.(type) {
        case *prometheus.CounterVec:
            counter, err := c.GetMetricWith(labels)
            if err != nil {
                log.Errorf("Failed to get counter metric: %v", err)
                continue
            }
            counter.Add(metricInstance.Value)
            
        case *prometheus.GaugeVec:
            gauge, err := c.GetMetricWith(labels)
            if err != nil {
                log.Errorf("Failed to get gauge metric: %v", err)
                continue
            }
            gauge.Set(metricInstance.Value)
            
        case *prometheus.HistogramVec:
            histogram, err := c.GetMetricWith(labels)
            if err != nil {
                log.Errorf("Failed to get histogram metric: %v", err)
                continue
            }
            histogram.Observe(metricInstance.Value)
        }
    }
    
    return nil
}

// 启动指标服务器
func (p *prometheusAdapter) startMetricsServer() error {
    mux := http.NewServeMux()
    mux.Handle("/metrics", promhttp.HandlerFor(p.registry, promhttp.HandlerOpts{}))
    
    p.server = &http.Server{
        Addr:    fmt.Sprintf(":%d", p.config.MetricsPort),
        Handler: mux,
    }
    
    go func() {
        if err := p.server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Errorf("Prometheus metrics server error: %v", err)
        }
    }()
    
    return nil
}

// 全局词汇表管理优化
type OptimizedWordList struct {
    // 全局词汇表
    globalWords []string
    
    // 词汇到索引的映射（优化查找性能）
    wordToIndex map[string]int32
    
    // 使用频率统计
    usageFreq map[string]int64
    
    // 读写锁保护
    rwMutex sync.RWMutex
}

// 智能词汇表更新
func (owl *OptimizedWordList) updateWordList() {
    owl.rwMutex.Lock()
    defer owl.rwMutex.Unlock()
    
    // 1. 按使用频率排序
    type wordFreq struct {
        word string
        freq int64
    }
    
    var wordFreqs []wordFreq
    for word, freq := range owl.usageFreq {
        wordFreqs = append(wordFreqs, wordFreq{word: word, freq: freq})
    }
    
    sort.Slice(wordFreqs, func(i, j int) bool {
        return wordFreqs[i].freq > wordFreqs[j].freq
    })
    
    // 2. 重建全局词汇表（高频词汇在前，提高查找性能）
    newGlobalWords := make([]string, len(wordFreqs))
    newWordToIndex := make(map[string]int32)
    
    for i, wf := range wordFreqs {
        newGlobalWords[i] = wf.word
        newWordToIndex[wf.word] = int32(i)
    }
    
    owl.globalWords = newGlobalWords
    owl.wordToIndex = newWordToIndex
}

// 优化的查找函数
func (owl *OptimizedWordList) fastLookup(index int32) (string, error) {
    owl.rwMutex.RLock()
    defer owl.rwMutex.RUnlock()
    
    if index < 0 || index >= int32(len(owl.globalWords)) {
        return "", fmt.Errorf("index %d out of range", index)
    }
    
    // 更新使用频率
    word := owl.globalWords[index]
    atomic.AddInt64(&owl.usageFreq[word], 1)
    
    return word, nil
}

// 批量属性处理优化
type BatchAttributeProcessor struct {
    // 批处理配置
    batchSize    int
    flushTimeout time.Duration
    
    // 处理队列
    queue chan *AttributeBatch
    
    // 工作池
    workerPool *WorkerPool
}

type AttributeBatch struct {
    Attributes []*mixerpb.CompressedAttributes
    Callback   func(results []ProcessResult)
    Timestamp  time.Time
}

// 批量处理工作器
func (bap *BatchAttributeProcessor) batchWorker(ctx context.Context) {
    timer := time.NewTimer(bap.flushTimeout)
    defer timer.Stop()
    
    var currentBatch []*AttributeBatch
    
    for {
        select {
        case batch := <-bap.queue:
            currentBatch = append(currentBatch, batch)
            
            // 检查是否达到批次大小
            if len(currentBatch) >= bap.batchSize {
                bap.processBatch(currentBatch)
                currentBatch = nil
                timer.Reset(bap.flushTimeout)
            }
            
        case <-timer.C:
            // 超时刷新
            if len(currentBatch) > 0 {
                bap.processBatch(currentBatch)
                currentBatch = nil
            }
            timer.Reset(bap.flushTimeout)
            
        case <-ctx.Done():
            // 处理剩余批次
            if len(currentBatch) > 0 {
                bap.processBatch(currentBatch)
            }
            return
        }
    }
}

// 处理单个批次
func (bap *BatchAttributeProcessor) processBatch(batches []*AttributeBatch) {
    // 1. 合并所有属性
    var allAttributes []*mixerpb.CompressedAttributes
    var callbacks []func(results []ProcessResult)
    
    for _, batch := range batches {
        allAttributes = append(allAttributes, batch.Attributes...)
        callbacks = append(callbacks, batch.Callback)
    }
    
    // 2. 并行处理属性
    results := make([]ProcessResult, len(allAttributes))
    
    // 使用工作池并行处理
    bap.workerPool.Process(allAttributes, func(i int, attr *mixerpb.CompressedAttributes) {
        bag := attribute.GetMutableBag(nil)
        defer bag.Done()
        
        if err := bag.UpdateBagFromProto(attr, bap.globalWordList); err != nil {
            results[i] = ProcessResult{Error: err}
            return
        }
        
        results[i] = ProcessResult{Bag: bag}
    })
    
    // 3. 调用回调函数
    for _, callback := range callbacks {
        if callback != nil {
            callback(results)
        }
    }
}