perf: 精排 P99 延迟优化 (180ms → ≤50ms)

algorithm/eas/easyrec_request.go

@@ -5,13 +5,21 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 
 	"github.com/alibaba/pairec/v2/algorithm/eas/easyrec"
 	"github.com/alibaba/pairec/v2/config"
 	"github.com/alibaba/pairec/v2/pkg/eas"
 	proto "github.com/golang/protobuf/proto"
 )
 
+// marshalBufPool reuses proto.Buffer to reduce marshal allocations
+var marshalBufPool = sync.Pool{
+	New: func() interface{} {
+		return proto.NewBuffer(make([]byte, 0, 4096))
+	},
+}
+
 type EasyrecRequest struct {
 	EasRequest
 	EasClient *eas.PredictClient
@@ -24,7 +32,14 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{},
 		return
 	}
 
-	data, _ := proto.Marshal(request)
+	buf := marshalBufPool.Get().(*proto.Buffer)
+	buf.Reset()
+	if err = buf.Marshal(request); err != nil {
+		marshalBufPool.Put(buf)
+		return
+	}
+	data := buf.Bytes()
+
 	if config.AppConfig.WarmUpData {
 		warmupFunc := func(data []byte) {
 			if file, err := os.OpenFile("warm_up.bin", os.O_CREATE|os.O_TRUNC|os.O_RDWR, 0664); err == nil {
@@ -36,6 +51,10 @@ func (r *EasyrecRequest) Invoke(requestData interface{}) (response interface{},
 	}
 
 	body, err := r.EasClient.BytesPredict(data)
+	// Only return normal-sized buffers to pool; let oversized ones be GC'd
+	if cap(buf.Bytes()) <= 1<<20 {
+		marshalBufPool.Put(buf)
+	}
 	if err != nil {
 		return
 	}

configloader.go

@@ -39,6 +39,7 @@ import (
 	"github.com/alibaba/pairec/v2/service/recall/berecall"
 	"github.com/alibaba/pairec/v2/sort"
 	"github.com/alibaba/pairec/v2/utils"
+	"github.com/alibaba/pairec/v2/web"
 )
 
 var (
@@ -118,6 +119,7 @@ func (l *ConfigLoader) reloadConfig(config *recconf.RecommendConfig) {
 	general_rank.LoadGeneralRankWithConfig(config)
 	pipeline.LoadPipelineConfigs(config)
 	fallback.LoadFallbackConfig(config)
+	initCallbackHandler(config)
 }
 
 func (l *ConfigLoader) loadConfigFromConfigServer() (*recconf.RecommendConfig, error) {
@@ -175,3 +177,19 @@ func ListenConfig(configName string) {
 
 	go loader.loopLoadConfig()
 }
+
+// initCallbackHandler initializes the callback worker pool from config.
+// Aggregates across all scenes: uses max WorkerPoolSize, OR of DropOnBackpressure.
+func initCallbackHandler(config *recconf.RecommendConfig) {
+	var poolSize int
+	var drop bool
+	for _, cbConf := range config.CallBackConfs {
+		if cbConf.WorkerPoolSize > poolSize {
+			poolSize = cbConf.WorkerPoolSize
+		}
+		if cbConf.DropOnBackpressure {
+			drop = true
+		}
+	}
+	web.InitHandler(poolSize, 5000, drop)
+}

go.mod

@@ -52,11 +52,14 @@ require (
 	github.com/google/uuid v1.3.0
 	github.com/mmcloughlin/geohash v0.10.0
 	go.uber.org/automaxprocs v1.5.2
+	golang.org/x/time v0.11.0
 )
 
 require (
 	github.com/alibabacloud-go/alibabacloud-gateway-spi v0.0.5 // indirect
 	github.com/alibabacloud-go/darabonba-openapi/v2 v2.1.13 // indirect
+	github.com/alibabacloud-go/endpoint-util v1.1.0 // indirect
+	github.com/alibabacloud-go/openapi-util v0.1.1 // indirect
 	github.com/alibabacloud-go/paifeaturestore-20230621/v6 v6.0.5 // indirect
 	github.com/alibabacloud-go/pairecservice-20221213/v3 v3.3.0 // indirect
 	github.com/alibabacloud-go/pairecservice-20221213/v4 v4.0.0 // indirect

go.sum

@@ -70,6 +70,7 @@ github.com/alibabacloud-go/tea-utils v1.3.1/go.mod h1:EI/o33aBfj3hETm4RLiAxF/ThQ
 github.com/alibabacloud-go/tea-utils v1.4.5 h1:h0/6Xd2f3bPE4XHTvkpjwxowIwRCJAJOqY6Eq8f3zfA=
 github.com/alibabacloud-go/tea-utils v1.4.5/go.mod h1:KNcT0oXlZZxOXINnZBs6YvgOd5aYp9U67G+E3R8fcQw=
 github.com/alibabacloud-go/tea-utils/v2 v2.0.5/go.mod h1:dL6vbUT35E4F4bFTHL845eUloqaerYBYPsdWR2/jhe4=
+github.com/alibabacloud-go/tea-utils/v2 v2.0.6/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I=
 github.com/alibabacloud-go/tea-utils/v2 v2.0.7 h1:WDx5qW3Xa5ZgJ1c8NfqJkF6w+AU5wB8835UdhPr6Ax0=
 github.com/alibabacloud-go/tea-utils/v2 v2.0.7/go.mod h1:qxn986l+q33J5VkialKMqT/TTs3E+U9MJpd001iWQ9I=
 github.com/aliyun/alibaba-cloud-sdk-go v1.63.92 h1:qespx4b6EexlXkvQUow9x0v1GnWUJYGU5FWYw3a4Wlg=
@@ -769,6 +770,8 @@ golang.org/x/text v0.30.0 h1:yznKA/E9zq54KzlzBEAWn1NXSQ8DIp/NYMy88xJjl4k=
 golang.org/x/text v0.30.0/go.mod h1:yDdHFIX9t+tORqspjENWgzaCVXgk0yYnYuSZ8UzzBVM=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
+golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

recconf/recconf.go

@@ -919,13 +919,17 @@ type AdjustCountConfig struct {
 	Weight     int
 }
 type CallBackConfig struct {
-	DataSource      DataSourceConfig
-	RankConf        RankConfig
-	RawFeatures     bool
-	RawFeaturesRate int
-	ItemSize        int
-	ItemSizeRate    int
-	UseUserFeatures bool
+	DataSource         DataSourceConfig
+	RankConf           RankConfig
+	RawFeatures        bool
+	RawFeaturesRate    int
+	ItemSize           int
+	ItemSizeRate       int
+	UseUserFeatures    bool
+	WorkerPoolSize     int     `json:"WorkerPoolSize"`
+	RateLimitQPS       float64 `json:"RateLimitQPS"`
+	JitterMaxMs        int     `json:"JitterMaxMs"`
+	DropOnBackpressure bool    `json:"DropOnBackpressure"`
 }
 type EmbeddingConfig struct {
 	DataSource DataSourceConfig

service/call_back_service.go

@@ -21,8 +21,28 @@ import (
 	"github.com/alibaba/pairec/v2/service/feature"
 	"github.com/alibaba/pairec/v2/service/rank"
 	"github.com/alibaba/pairec/v2/utils"
+	"golang.org/x/time/rate"
 )
 
+var callbackLimiters sync.Map
+
+func getCallbackLimiter(scene string, qps float64) *rate.Limiter {
+	burst := int(qps)
+	if burst < 1 {
+		burst = 1
+	}
+	if v, ok := callbackLimiters.Load(scene); ok {
+		limiter := v.(*rate.Limiter)
+		// Support hot reload: update limit if config changed
+		limiter.SetLimit(rate.Limit(qps))
+		limiter.SetBurst(burst)
+		return limiter
+	}
+	limiter := rate.NewLimiter(rate.Limit(qps), burst)
+	actual, _ := callbackLimiters.LoadOrStore(scene, limiter)
+	return actual.(*rate.Limiter)
+}
+
 type CallBackService struct {
 	recallService      *RecallService
 	featureService     *feature.FeatureService
@@ -239,6 +259,16 @@ func (r *CallBackService) Rank(context *context.RecommendContext) {
 					}
 				}
 			}
+			if callBackConfig.RateLimitQPS > 0 {
+				limiter := getCallbackLimiter(scene_name, callBackConfig.RateLimitQPS)
+				if !limiter.Allow() {
+					log.Warning(fmt.Sprintf("requestId=%s\tcallback rate limited, skipping EAS call", context.RecommendId))
+					return
+				}
+			}
+			if callBackConfig.JitterMaxMs > 0 {
+				time.Sleep(time.Duration(rand.Intn(callBackConfig.JitterMaxMs)) * time.Millisecond)
+			}
 			// run 返回原始的值，然后处理返回数据// 注册配置
 			ret, err := algorithm.Run(newAlgoName, algoData.GetFeatures())
 			if err != nil {

service/metrics/metrics.go

@@ -11,24 +11,28 @@ import (
 )
 
 var (
-	SizeNotEnoughTotal    *prometheus.CounterVec
-	RecTotal              *prometheus.CounterVec
-	RecallCount           *prometheus.CounterVec
-	RecallCountTotal      *prometheus.CounterVec
-	RecallItemsPercentage *prometheus.GaugeVec
-	RecallDurSecs         *prometheus.HistogramVec
-	FilterDurSecs         *prometheus.HistogramVec
-	GeneralRankDurSecs    *prometheus.HistogramVec
-	LoadFeatureDurSecs    *prometheus.HistogramVec
-	RankDurSecs           *prometheus.HistogramVec
-	SortDurSecs           *prometheus.HistogramVec
-	RecDurSecs            *prometheus.HistogramVec
-	FallbackTotal         *prometheus.CounterVec
+	SizeNotEnoughTotal      *prometheus.CounterVec
+	RecTotal                *prometheus.CounterVec
+	RecallCount             *prometheus.CounterVec
+	RecallCountTotal        *prometheus.CounterVec
+	RecallItemsPercentage   *prometheus.GaugeVec
+	RecallDurSecs           *prometheus.HistogramVec
+	FilterDurSecs           *prometheus.HistogramVec
+	GeneralRankDurSecs      *prometheus.HistogramVec
+	LoadFeatureDurSecs      *prometheus.HistogramVec
+	RankDurSecs             *prometheus.HistogramVec
+	SortDurSecs             *prometheus.HistogramVec
+	RecDurSecs              *prometheus.HistogramVec
+	FallbackTotal           *prometheus.CounterVec
+	CallbackChannelPending  prometheus.GaugeFunc
 
 	enabled = false
 	once    sync.Once
 )
 
+// CallbackPendingFunc is set by the web package to avoid import cycle
+var CallbackPendingFunc func() int64
+
 var CustomRegister = prometheus.NewRegistry()
 
 func Enabled() bool {
@@ -51,7 +55,8 @@ func Load(conf *recconf.RecommendConfig) {
 			LoadFeatureDurSecs,
 			RankDurSecs,
 			SortDurSecs,
-			FallbackTotal)
+			FallbackTotal,
+			CallbackChannelPending)
 	})
 
 	enabled = conf.PrometheusConfig.Enable
@@ -173,4 +178,15 @@ func initMetrics(conf *recconf.RecommendConfig) {
 		Name:      "fallback_total",
 		Help:      "How many times of fallback recommend.",
 	}, []string{"scene"})
+
+	CallbackChannelPending = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
+		Subsystem: subsystem,
+		Name:      "callback_channel_pending",
+		Help:      "Number of pending callback requests in channel.",
+	}, func() float64 {
+		if CallbackPendingFunc != nil {
+			return float64(CallbackPendingFunc())
+		}
+		return 0
+	})
 }

service/rank/algo_data.go

@@ -86,8 +86,9 @@ func (d *EasyrecAlgoData) GetFeatures() interface{} {
 }
 
 type AlgoDataGenerator struct {
-	requestData []map[string]interface{}
-	requestItem []*module.Item
+	requestData  []map[string]interface{}
+	requestItem  []*module.Item
+	userFeatures map[string]interface{}
 }
 
 // SetItemFeatures implements IAlgoDataGenerator.
@@ -105,22 +106,31 @@ func (g *AlgoDataGenerator) AddFeatures(item *module.Item, itemFeatures map[stri
 	if item != nil {
 		g.requestItem = append(g.requestItem, item)
 	}
-
-	features := make(map[string]interface{}, len(itemFeatures)+len(userFeatures))
-	for k, v := range userFeatures {
-		features[k] = v
+	if g.userFeatures == nil {
+		g.userFeatures = userFeatures
 	}
+
+	features := make(map[string]interface{}, len(itemFeatures))
 	for k, v := range itemFeatures {
 		features[k] = v
 	}
-
 	g.requestData = append(g.requestData, features)
 }
 func (g *AlgoDataGenerator) GeneratorAlgoData() IAlgoData {
 	copydata := make([]map[string]interface{}, len(g.requestData))
+	for i, itemFeatures := range g.requestData {
+		merged := make(map[string]interface{}, len(itemFeatures)+len(g.userFeatures))
+		for k, v := range g.userFeatures {
+			merged[k] = v
+		}
+		for k, v := range itemFeatures {
+			merged[k] = v
+		}
+		copydata[i] = merged
+	}
+
 	copyItems := make([]*module.Item, len(g.requestItem))
 	copy(copyItems, g.requestItem)
-	copy(copydata, g.requestData)
 
 	algoData := &AlgoData{
 		AlgoDataBase: &AlgoDataBase{
@@ -183,8 +193,8 @@ type EasyrecAlgoDataGenerator struct {
 
 func NewEasyrecAlgoDataGenerator(contextFeatures []string) *EasyrecAlgoDataGenerator {
 	generator := &EasyrecAlgoDataGenerator{
-		requestItem:     make([]*module.Item, 0, 100),
-		contextFeatures: make(map[string][]interface{}, 8),
+		requestItem:     make([]*module.Item, 0, 128),
+		contextFeatures: make(map[string][]interface{}, len(contextFeatures)+8),
 		parseFeature:    true,
 	}
 
@@ -195,6 +205,7 @@ func NewEasyrecAlgoDataGenerator(contextFeatures []string) *EasyrecAlgoDataGener
 				valueType: reflect.TypeOf(""),
 			}
 			generator.itemFeatures = append(generator.itemFeatures, feature)
+			generator.contextFeatures[featureName] = make([]interface{}, 0, 128)
 		}
 	}
 
@@ -274,14 +285,15 @@ func (g *EasyrecAlgoDataGenerator) AddFeatures(item *module.Item, itemFeatures m
 }
 
 func (g *EasyrecAlgoDataGenerator) GeneratorAlgoData() IAlgoData {
-	copyItems := make([]*module.Item, len(g.requestItem))
-	copy(copyItems, g.requestItem)
+	// Transfer ownership instead of copy — generator resets immediately after
+	items := g.requestItem
+	g.requestItem = make([]*module.Item, 0, 128)
 
 	builder := easyrec.NewEasyrecRequestBuilder()
 	for k, v := range g.userFeatures {
 		builder.AddUserFeature(k, v)
 	}
-	for _, item := range g.requestItem {
+	for _, item := range items {
 		builder.AddItemId(string(item.Id))
 	}
 	for k, v := range g.contextFeatures {
@@ -295,13 +307,12 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoData() IAlgoData {
 
 	algoData := &EasyrecAlgoData{
 		AlgoDataBase: &AlgoDataBase{
-			Items:      copyItems,
+			Items:      items,
 			AlgoResult: make(map[string][]response.AlgoResponse),
 		},
 		easyrecRequest: builder.EasyrecRequest(),
 	}
 
-	g.requestItem = g.requestItem[:0]
 	return algoData
 }
 
@@ -310,8 +321,9 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebug() IAlgoData {
 }
 
 func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, meta map[string]string) IAlgoData {
-	copyItems := make([]*module.Item, len(g.requestItem))
-	copy(copyItems, g.requestItem)
+	// Transfer ownership instead of copy — generator resets immediately after
+	items := g.requestItem
+	g.requestItem = make([]*module.Item, 0, 128)
 
 	builder := easyrec.NewEasyrecRequestBuilderDebugWithLevel(level)
 	if meta != nil && len(meta) > 0 {
@@ -321,7 +333,7 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, me
 	for k, v := range g.userFeatures {
 		builder.AddUserFeature(k, v)
 	}
-	for _, item := range g.requestItem {
+	for _, item := range items {
 		builder.AddItemId(string(item.Id))
 	}
 	for k, v := range g.contextFeatures {
@@ -335,13 +347,12 @@ func (g *EasyrecAlgoDataGenerator) GeneratorAlgoDataDebugWithLevel(level int, me
 
 	algoData := &EasyrecAlgoData{
 		AlgoDataBase: &AlgoDataBase{
-			Items:      copyItems,
+			Items:      items,
 			AlgoResult: make(map[string][]response.AlgoResponse),
 		},
 		easyrecRequest: builder.EasyrecRequest(),
 	}
 
-	g.requestItem = g.requestItem[:0]
 	return algoData
 }