Spring Cloud+AI ：实现分布式智能推荐系统

欢迎文末添加好友交流，共同进步！

“ 俺はモンキー・D・ルフィ。海贼王になる男だ！”

引言

• 在当今数字化时代，推荐系统已成为电商平台、内容分发平台、社交网络等互联网产品的核心竞争力之一。从淘宝的"猜你喜欢"、抖音的精准内容推送，到 Netflix 的影视推荐，优秀的推荐系统不仅能显著提升用户留存率和转化率，更能为企业带来可观的商业价值。据统计，亚马逊约 35% 的销售额来自推荐系统，Netflix 则通过推荐算法为用户节省了每年约 10 亿美元的搜索成本。
• 然而，随着业务规模的增长和推荐算法的复杂化，传统的单体架构逐渐暴露出诸多瓶颈。首先，推荐系统涉及用户画像构建、实时行为收集、特征工程、模型推理等多个环节，单体应用难以应对日益复杂的业务逻辑；其次，推荐服务需要处理海量并发请求，单机部署无法满足弹性伸缩的需求；再者，AI 模型的迭代更新日益频繁，单体架构下模型部署往往需要重启整个应用，严重影响线上服务稳定性；最后，企业需要支持 A/B 测试以验证不同算法策略的效果，单体架构难以实现灵活的流量隔离和策略切换。
• 基于上述挑战，采用 Spring Cloud 微服务架构结合分布式 AI 服务成为企业级推荐系统的主流解决方案。这种架构实现了计算与存储的解耦、特征工程与模型推理的分离，支持按需水平扩展，能够轻松应对流量洪峰。同时，微服务架构天然支持模型热更新和灰度发布，使算法迭代更加敏捷。本文将基于 Spring Boot 3 和 Spring Cloud 2022，详细介绍如何从零构建一个高可用、可扩展的智能推荐系统，重点探讨如何在 Java 生态中安全、高效地集成 AI 能力。

整体架构设计

一个完整的推荐系统需要处理从用户请求到推荐结果返回的全链路流程。在微服务架构下，我们将整个系统拆分为多个职责单一的服务，通过服务间协作完成推荐任务。核心服务包括用户服务、商品服务、特征工程服务、推荐服务和模型推理服务。

服务划分与职责

• user-service：负责用户基础信息和画像数据的维护，提供用户注册、登录、画像查询等接口。用户画像包括性别、年龄、地域、会员等级等静态属性，以及历史偏好标签等动态属性。
• item-service：管理商品/内容元数据，包括商品分类、品牌、价格、库存、标签等信息。同时维护商品特征向量，用于相似度计算。
• recommendation-service：推荐系统的核心服务，负责协调其他服务完成推荐流程。它接收用户请求，调用用户和商品服务获取上下文，通过特征工程服务组装特征向量，最终调用模型服务获取推荐结果。
• feature-engine：特征工程服务，负责实时特征提取和处理。包括用户实时行为特征（浏览、点击、购买）、商品热度特征、上下文特征（时间、设备、地理位置）等。
• model-serving：模型推理服务，负责加载训练好的推荐模型并提供推理接口。该服务通常使用 Python 实现，利用 PyTorch/TensorFlow 的推理能力，通过 REST 或 gRPC 暴露服务。
• event-collector：事件收集服务，负责收集用户曝光、点击、转化等行为数据，并将这些数据发送到消息队列（如 Kafka），用于后续的离线训练和在线学习。

基础设施组件

• 服务注册与发现：使用 Nacos 或 Eureka 作为注册中心，实现服务的动态注册与发现，支持服务健康检查和故障剔除。
• API 网关：使用 Spring Cloud Gateway 作为统一入口，负责路由转发、认证授权、限流熔断、协议转换等功能。
• 配置中心：使用 Nacos Config 或 Spring Cloud Config 集中管理各服务的配置文件，支持配置的动态刷新。
• 消息队列：使用 Kafka 或 RocketMQ 处理异步事件流，实现实时日志收集和在线学习。
• 缓存层：使用 Redis 缓存热点推荐结果和特征数据，减少数据库和模型服务的压力。

系统调用关系

1┌─────────────────────────────────────────────────────────────────────┐
2│                         API Gateway (Spring Cloud Gateway)           │
3│                    认证 | 限流 | 路由 | 熔断                           │
4└─────────────────────────────────────────────────────────────────────┘
5                                    │
6                                    ▼
7┌─────────────────────────────────────────────────────────────────────┐
8│                    Recommendation Service (推荐服务)                   │
9│              ┌─────────────────────────────────────────────┐         │
10│              │  Controller → Service → Feature Assembler    │         │
11│              └─────────────────────────────────────────────┘         │
12└─────┬───────────────┬───────────────┬───────────────┬───────────────┘
13      │               │               │               │
14      ▼               ▼               ▼               ▼
15┌───────────┐  ┌───────────┐  ┌───────────┐  ┌─────────────────────────┐
16│   User    │  │   Item    │  │  Feature  │  │     Model Serving        │
17│  Service  │  │  Service  │  │  Engine   │  │   (Python/PyTorch)       │
18│           │  │           │  │           │  │  • 协同过滤              │
19│ • 用户信息 │  │ • 商品信息 │  │ • 实时行为 │  │  • 矩阵分解              │
20│ • 用户画像 │  │ • 商品特征 │  │ • 特征组装 │  │  • NCF/DeepFM           │
21└─────┬─────┘  └─────┬─────┘  └─────┬─────┘  └─────────┬───────────────┘
22      │               │               │                    │
23      └───────────────┴───────────────┴────────────────────┘
24                          │
25                          ▼
26              ┌───────────────────────────┐
27              │    Nacos (注册中心)        │
28              │  • 服务注册与发现           │
29              │  • 配置中心                │
30              │  • 健康检查                │
31              └───────────────────────────┘
32
33                          │
34                          ▼
35              ┌───────────────────────────┐
36              │  Kafka (消息队列)          │
37              │  • 曝光事件                │
38              │  • 点击事件                │
39              │  • 转化事件                │
40              └───────────────────────────┘
41                          │
42                          ▼
43              ┌───────────────────────────┐
44              │  Event Collector Service   │
45              │  • 实时日志收集            │
46              │  • 在线学习反馈            │
47              └───────────────────────────┘
48

架构设计亮点

1. 特征与模型解耦：特征工程独立为微服务，模型推理独立为 Python 服务，两者通过标准化接口交互。这种设计使特征工程和模型开发可以并行迭代，互不影响。
2. 水平扩展能力：各服务可根据负载独立扩容，如双11大促期间可以临时增加推荐服务和模型服务的实例数，应对流量洪峰。
3. 容错与降级：通过 Resilience4j 实现服务熔断和降级，当模型服务不可用时，可以降级到基于规则的推荐（如热门商品推荐），确保核心业务不受影响。
4. 灰度发布支持：通过网关的流量路由能力，可以实现不同算法版本的灰度测试，验证新算法效果后再全量发布。

AI 模型选型与训练

推荐算法的选择需要综合考虑业务场景、数据规模、实时性要求和计算资源。对于电商和内容平台，常见的推荐算法包括基于协同过滤的传统方法、基于矩阵分解的隐因子模型，以及基于深度学习的神经网络模型。

算法选型分析

1. 协同过滤（Collaborative Filtering）是最经典的推荐算法之一，分为基于用户和基于物品两种。用户协同过滤假设相似用户有相似偏好，通过找到与目标用户相似的用户群，推荐这些用户喜欢的物品。物品协同过滤则基于物品间的相似度，推荐与用户历史行为中物品相似的其他物品。协同过滤实现简单、可解释性强，但存在冷启动和稀疏性问题。
2. 矩阵分解（Matrix Factorization）通过将用户-物品评分矩阵分解为低维的用户矩阵和物品矩阵，捕捉潜在因子关系。SVD（奇异值分解）和其变体 SVD++ 是其中的代表算法。矩阵分解相比协同过滤能更好地处理数据稀疏问题，是目前工业界应用最广泛的算法之一。
3. 深度学习模型如 NCF（Neural Collaborative Filtering）、DeepFM、Wide&Deep 等能够捕捉高阶特征交互，在效果上通常优于传统方法。NCF 用神经网络替代矩阵分解的内积操作，增强表达能力；DeepFM 结合了因子分解机和深度网络的优势；Wide&Deep 则通过联合训练线性模型和深度模型，平衡记忆与泛化能力。

模型训练与导出

在实际项目中，我们通常使用 Python 生态中的 PyTorch 或 TensorFlow 进行模型训练。以下是一个使用 PyTorch 实现 NCF 模型的简化示例：

1# train_ncf_model.py
2import torch
3import torch.nn as nn
4import numpy as np
5from torch.utils.data import Dataset, DataLoader
6
7# 定义 NCF 模型
8class NCFModel(nn.Module):
9    def __init__(self, num_users, num_items, embedding_dim=32):
10        super(NCFModel, self).__init__()
11        # 用户嵌入层
12        self.user_embedding = nn.Embedding(num_users, embedding_dim)
13        # 物品嵌入层
14        self.item_embedding = nn.Embedding(num_items, embedding_dim)
15        # MLP 层
16        self.mlp = nn.Sequential(
17            nn.Linear(embedding_dim * 2, 128),
18            nn.ReLU(),
19            nn.Dropout(0.2),
20            nn.Linear(128, 64),
21            nn.ReLU(),
22            nn.Dropout(0.2),
23            nn.Linear(64, 1),
24            nn.Sigmoid()
25        )
26
27    def forward(self, user_ids, item_ids):
28        user_emb = self.user_embedding(user_ids)
29        item_emb = self.item_embedding(item_ids)
30        concat = torch.cat([user_emb, item_emb], dim=-1)
31        return self.mlp(concat)
32
33# 自定义数据集
34class RecommendationDataset(Dataset):
35    def __init__(self, user_ids, item_ids, labels):
36        self.user_ids = torch.LongTensor(user_ids)
37        self.item_ids = torch.LongTensor(item_ids)
38        self.labels = torch.FloatTensor(labels)
39
40    def __len__(self):
41        return len(self.labels)
42
43    def __getitem__(self, idx):
44        return self.user_ids[idx], self.item_ids[idx], self.labels[idx]
45
46# 训练函数
47def train_model(train_data, val_data, num_users, num_items, epochs=10):
48    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
49    model = NCFModel(num_users, num_items).to(device)
50    criterion = nn.BCELoss()
51    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
52
53    train_loader = DataLoader(train_data, batch_size=256, shuffle=True)
54
55    for epoch in range(epochs):
56        model.train()
57        total_loss = 0
58        for user_ids, item_ids, labels in train_loader:
59            user_ids, item_ids, labels = user_ids.to(device), item_ids.to(device), labels.to(device)
60
61            optimizer.zero_grad()
62            predictions = model(user_ids, item_ids).squeeze()
63            loss = criterion(predictions, labels)
64            loss.backward()
65            optimizer.step()
66
67            total_loss += loss.item()
68
69        print(f'Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}')
70
71    return model
72
73# 导出模型为 ONNX 格式
74def export_to_onnx(model, output_path='ncf_model.onnx'):
75    model.eval()
76    dummy_user_ids = torch.LongTensor([0])
77    dummy_item_ids = torch.LongTensor([0])
78
79    torch.onnx.export(
80        model,
81        (dummy_user_ids, dummy_item_ids),
82        output_path,
83        input_names=['user_ids', 'item_ids'],
84        output_names=['prediction'],
85        dynamic_axes={
86            'user_ids': {0: 'batch_size'},
87            'item_ids': {0: 'batch_size'},
88            'prediction': {0: 'batch_size'}
89        },
90        opset_version=14
91    )
92    print(f'Model exported to {output_path}')
93
94if __name__ == '__main__':
95    # 模拟训练数据
96    num_users = 10000
97    num_items = 50000
98    num_samples = 100000
99
100    user_ids = np.random.randint(0, num_users, num_samples)
101    item_ids = np.random.randint(0, num_items, num_samples)
102    labels = np.random.randint(0, 2, num_samples).astype(float)
103
104    # 划分训练集和验证集
105    split_idx = int(0.8 * num_samples)
106    train_data = RecommendationDataset(user_ids[:split_idx], item_ids[:split_idx], labels[:split_idx])
107    val_data = RecommendationDataset(user_ids[split_idx:], item_ids[split_idx:], labels[split_idx:])
108
109    # 训练模型
110    model = train_model(train_data, val_data, num_users, num_items, epochs=5)
111
112    # 导出为 ONNX 格式
113    export_to_onnx(model)
114

模型服务化

训练完成后，我们需要将模型部署为服务供 Java 应用调用。以下是使用 FastAPI 实现的模型推理服务：

1# model_server.py
2from fastapi import FastAPI, HTTPException
3from pydantic import BaseModel
4import torch
5import onnxruntime as ort
6import numpy as np
7from typing import List
8
9app = FastAPI(title='Recommendation Model API')
10
11# 加载 ONNX 模型
12session = ort.InferenceSession('ncf_model.onnx')
13
14class PredictionRequest(BaseModel):
15    user_ids: List[int]
16    item_ids: List[int]
17
18class PredictionResponse(BaseModel):
19    predictions: List[float]
20
21@app.post('/api/v1/predict', response_model=PredictionResponse)
22async def predict(request: PredictionRequest):
23    try:
24        # 准备输入数据
25        user_ids = np.array(request.user_ids, dtype=np.int64).reshape(-1, 1)
26        item_ids = np.array(request.item_ids, dtype=np.int64).reshape(-1, 1)
27
28        # ONNX 推理
29        inputs = {
30            'user_ids': user_ids,
31            'item_ids': item_ids
32        }
33        predictions = session.run(None, inputs)[0]
34
35        return PredictionResponse(predictions=predictions.tolist())
36
37    except Exception as e:
38        raise HTTPException(status_code=500, detail=str(e))
39
40@app.get('/health')
41async def health_check():
42    return {'status': 'healthy'}
43
44if __name__ == '__main__':
45    import uvicorn
46    uvicorn.run(app, host='0.0.0.0', port=8000)
47

这个模型服务提供了 REST API，Java 应用可以通过 HTTP 调用。对于高性能场景，也可以考虑使用 gRPC 协议，它比 REST 更高效，支持双向流式通信。

核心微服务实现

本节将详细介绍推荐系统中核心微服务的实现，包括服务间调用、特征组装、模型推理集成、异步日志收集和熔断降级等关键功能。所有代码基于 Spring Boot 3 和 Spring Cloud 2022。

项目结构与依赖

首先，创建一个 Maven 多模块项目：

1<!-- pom.xml -->
2<project>
3    <groupId>com.example.recommendation</groupId>
4    <artifactId>recommendation-system</artifactId>
5    <version>1.0.0</version>
6    <packaging>pom</packaging>
7
8    <modules>
9        <module>user-service</module>
10        <module>item-service</module>
11        <module>recommendation-service</module>
12        <module>feature-engine</module>
13        <module>common</module>
14    </modules>
15
16    <properties>
17        <java.version>17</java.version>
18        <spring-boot.version>3.2.0</spring-boot.version>
19        <spring-cloud.version>2023.0.0</spring-cloud.version>
20    </properties>
21
22    <dependencyManagement>
23        <dependencies>
24            <dependency>
25                <groupId>org.springframework.boot</groupId>
26                <artifactId>spring-boot-dependencies</artifactId>
27                <version>${spring-boot.version}</version>
28                <type>pom</type>
29                <scope>import</scope>
30            </dependency>
31            <dependency>
32                <groupId>org.springframework.cloud</groupId>
33                <artifactId>spring-cloud-dependencies</artifactId>
34                <version>${spring-cloud.version}</version>
35                <type>pom</type>
36                <scope>import</scope>
37            </dependency>
38        </dependencies>
39    </dependencyManagement>
40</project>
41

User Service 实现

用户服务提供用户基础信息和画像查询接口：

1// user-service/src/main/java/com/example/user/controller/UserController.java
2package com.example.user.controller;
3
4import com.example.user.dto.UserProfileDTO;
5import com.example.user.service.UserService;
6import lombok.RequiredArgsConstructor;
7import org.springframework.web.bind.annotation.*;
8
9@RestController
10@RequestMapping("/api/users")
11@RequiredArgsConstructor
12public class UserController {
13
14    private final UserService userService;
15
16    /**
17     * 获取用户画像
18     */
19    @GetMapping("/{userId}/profile")
20    public UserProfileDTO getUserProfile(@PathVariable Long userId) {
21        return userService.getUserProfile(userId);
22    }
23
24    /**
25     * 批量获取用户画像
26     */
27    @PostMapping("/profiles/batch")
28    public Map<Long, UserProfileDTO> getUserProfilesBatch(@RequestBody List<Long> userIds) {
29        return userService.getUserProfilesBatch(userIds);
30    }
31}
32

1// user-service/src/main/java/com/example/user/service/UserService.java
2package com.example.user.service;
3
4import com.example.user.dto.UserProfileDTO;
5import com.example.user.entity.UserProfile;
6import com.example.user.repository.UserProfileRepository;
7import lombok.RequiredArgsConstructor;
8import org.springframework.cache.annotation.Cacheable;
9import org.springframework.stereotype.Service;
10
11import java.util.List;
12import java.util.Map;
13import java.util.stream.Collectors;
14
15@Service
16@RequiredArgsConstructor
17public class UserService {
18
19    private final UserProfileRepository userProfileRepository;
20
21    /**
22     * 获取用户画像，使用 Redis 缓存
23     */
24    @Cacheable(value = "userProfiles", key = "#userId")
25    public UserProfileDTO getUserProfile(Long userId) {
26        UserProfile profile = userProfileRepository.findById(userId)
27                .orElseThrow(() -> new RuntimeException("User not found: " + userId));
28
29        return UserProfileDTO.builder()
30                .userId(profile.getUserId())
31                .gender(profile.getGender())
32                .age(profile.getAge())
33                .city(profile.getCity())
34                .membershipLevel(profile.getMembershipLevel())
35                .interestTags(profile.getInterestTags())
36                .build();
37    }
38
39    /**
40     * 批量获取用户画像
41     */
42    public Map<Long, UserProfileDTO> getUserProfilesBatch(List<Long> userIds) {
43        List<UserProfile> profiles = userProfileRepository.findAllById(userIds);
44
45        return profiles.stream()
46                .collect(Collectors.toMap(
47                        UserProfile::getUserId,
48                        profile -> UserProfileDTO.builder()
49                                .userId(profile.getUserId())
50                                .gender(profile.getGender())
51                                .age(profile.getAge())
52                                .city(profile.getCity())
53                                .membershipLevel(profile.getMembershipLevel())
54                                .interestTags(profile.getInterestTags())
55                                .build()
56                ));
57    }
58}
59

1# user-service/src/main/resources/application.yml
2server:
3  port: 8081
4
5spring:
6  application:
7    name: user-service
8  datasource:
9    url: jdbc:mysql://localhost:3306/user_db
10    username: root
11    password: password
12  data:
13    redis:
14      host: localhost
15      port: 6379
16
17# Nacos 注册中心配置
18spring.cloud.nacos:
19  discovery:
20    server-addr: localhost:8848
21    namespace: public
22  config:
23    server-addr: localhost:8848
24    file-extension: yml
25

Recommendation Service 实现

推荐服务是整个系统的核心，负责协调各服务完成推荐流程。首先定义 Feign 客户端：

1// recommendation-service/src/main/java/com/example/recommendation/client/UserClient.java
2package com.example.recommendation.client;
3
4import com.example.common.dto.UserProfileDTO;
5import org.springframework.cloud.openfeign.FeignClient;
6import org.springframework.web.bind.annotation.GetMapping;
7import org.springframework.web.bind.annotation.PathVariable;
8import org.springframework.web.bind.annotation.PostMapping;
9import org.springframework.web.bind.annotation.RequestBody;
10
11import java.util.List;
12import java.util.Map;
13
14@FeignClient(name = "user-service", path = "/api/users", fallbackFactory = UserClientFallback.class)
15public interface UserClient {
16
17    @GetMapping("/{userId}/profile")
18    UserProfileDTO getUserProfile(@PathVariable("userId") Long userId);
19
20    @PostMapping("/profiles/batch")
21    Map<Long, UserProfileDTO> getUserProfilesBatch(@RequestBody List<Long> userIds);
22}
23

1// recommendation-service/src/main/java/com/example/recommendation/client/ItemClient.java
2package com.example.recommendation.client;
3
4import com.example.common.dto.ItemDTO;
5import org.springframework.cloud.openfeign.FeignClient;
6import org.springframework.web.bind.annotation.GetMapping;
7import org.springframework.web.bind.annotation.PathVariable;
8import org.springframework.web.bind.annotation.PostMapping;
9import org.springframework.web.bind.annotation.RequestBody;
10
11import java.util.List;
12import java.util.Map;
13
14@FeignClient(name = "item-service", path = "/api/items", fallbackFactory = ItemClientFallback.class)
15public interface ItemClient {
16
17    @GetMapping("/{itemId}")
18    ItemDTO getItem(@PathVariable("itemId") Long itemId);
19
20    @PostMapping("/batch")
21    List<ItemDTO> getItemsBatch(@RequestBody List<Long> itemIds);
22
23    @GetMapping("/category/{category}")
24    List<ItemDTO> getItemsByCategory(@PathVariable("category") String category);
25}
26

实现降级工厂类：

1// recommendation-service/src/main/java/com/example/recommendation/client/UserClientFallback.java
2package com.example.recommendation.client;
3
4import com.example.common.dto.UserProfileDTO;
5import lombok.extern.slf4j.Slf4j;
6import org.springframework.stereotype.Component;
7
8import java.util.Collections;
9import java.util.List;
10import java.util.Map;
11
12@Slf4j
13@Component
14public class UserClientFallback implements UserClient {
15
16    @Override
17    public UserProfileDTO getUserProfile(Long userId) {
18        log.warn("User service fallback triggered for userId: {}", userId);
19        // 返回默认用户画像
20        return UserProfileDTO.builder()
21                .userId(userId)
22                .gender("unknown")
23                .age(25)
24                .city("unknown")
25                .membershipLevel("NORMAL")
26                .interestTags(Collections.emptyList())
27                .build();
28    }
29
30    @Override
31    public Map<Long, UserProfileDTO> getUserProfilesBatch(List<Long> userIds) {
32        log.warn("User service batch fallback triggered");
33        return Collections.emptyMap();
34    }
35}
36

推荐服务核心逻辑：

1// recommendation-service/src/main/java/com/example/recommendation/service/RecommendationService.java
2package com.example.recommendation.service;
3
4import com.example.common.dto.*;
5import com.example.recommendation.client.ItemClient;
6import com.example.recommendation.client.UserClient;
7import com.example.recommendation.client.ModelClient;
8import com.example.recommendation.config.RecommendationProperties;
9import lombok.RequiredArgsConstructor;
10import lombok.extern.slf4j.Slf4j;
11import org.springframework.data.redis.core.RedisTemplate;
12import org.springframework.stereotype.Service;
13
14import java.time.Duration;
15import java.util.*;
16import java.util.stream.Collectors;
17
18@Slf4j
19@Service
20@RequiredArgsConstructor
21public class RecommendationService {
22
23    private final UserClient userClient;
24    private final ItemClient itemClient;
25    private final ModelClient modelClient;
26    private final FeatureEngineClient featureEngineClient;
27    private final RedisTemplate<String, Object> redisTemplate;
28    private final KafkaTemplate<String, Object> kafkaTemplate;
29    private final RecommendationProperties properties;
30
31    /**
32     * 获取个性化推荐
33     */
34    public RecommendationResult getRecommendations(Long userId, String scenario, int size) {
35        // 1. 尝试从缓存获取
36        String cacheKey = String.format("recommendation:%d:%s", userId, scenario);
37        List<RecommendedItem> cachedResult = (List<RecommendedItem>) redisTemplate.opsForValue().get(cacheKey);
38
39        if (cachedResult != null) {
40            log.info("Cache hit for user: {}, scenario: {}", userId, scenario);
41            return RecommendationResult.builder()
42                    .userId(userId)
43                    .scenario(scenario)
44                    .items(cachedResult)
45                    .source("CACHE")
46                    .build();
47        }
48
49        // 2. 获取用户画像
50        UserProfileDTO userProfile = userClient.getUserProfile(userId);
51
52        // 3. 获取候选商品集（从召回池中筛选）
53        List<Long> candidateItemIds = getCandidateItems(userProfile, scenario, size * 10);
54
55        // 4. 批量获取商品信息
56        List<ItemDTO> candidateItems = itemClient.getItemsBatch(candidateItemIds);
57
58        // 5. 特征工程
59        FeatureVector featureVector = buildFeatureVector(userId, userProfile, candidateItems);
60
61        // 6. 模型推理评分
62        Map<Long, Double> itemScores = modelClient.predict(userId, candidateItemIds, featureVector);
63
64        // 7. 排序并返回 Top-K
65        List<RecommendedItem> recommendedItems = candidateItems.stream()
66                .filter(item -> itemScores.containsKey(item.getItemId()))
67                .sorted((a, b) -> Double.compare(
68                        itemScores.get(b.getItemId()),
69                        itemScores.get(a.getItemId())
70                ))
71                .limit(size)
72                .map(item -> RecommendedItem.builder()
73                        .itemId(item.getItemId())
74                        .itemName(item.getItemName())
75                        .category(item.getCategory())
76                        .price(item.getPrice())
77                        .score(itemScores.get(item.getItemId()))
78                        .reason("基于您的兴趣推荐")
79                        .build())
80                .collect(Collectors.toList());
81
82        // 8. 缓存结果
83        redisTemplate.opsForValue().set(
84                cacheKey,
85                recommendedItems,
86                Duration.ofMinutes(properties.getCacheExpireMinutes())
87        );
88
89        // 9. 异步记录曝光事件
90        recordExposureEvent(userId, scenario, recommendedItems);
91
92        return RecommendationResult.builder()
93                .userId(userId)
94                .scenario(scenario)
95                .items(recommendedItems)
96                .source("MODEL")
97                .build();
98    }
99
100    /**
101     * 获取候选商品集
102     */
103    private List<Long> getCandidateItems(UserProfileDTO userProfile, String scenario, int poolSize) {
104        // 实际实现中，这里可以从离线计算的召回池中获取
105        // 这里简化为按用户兴趣标签获取相关商品
106        List<String> interests = userProfile.getInterestTags();
107        if (interests.isEmpty()) {
108            interests = Arrays.asList("热门");
109        }
110
111        return itemClient.getItemsByCategory(interests.get(0)).stream()
112                .limit(poolSize)
113                .map(ItemDTO::getItemId)
114                .collect(Collectors.toList());
115    }
116
117    /**
118     * 构建特征向量
119     */
120    private FeatureVector buildFeatureVector(Long userId, UserProfileDTO userProfile, List<ItemDTO> items) {
121        return FeatureVector.builder()
122                .userId(userId)
123                .userGender(userProfile.getGender())
124                .userAge(userProfile.getAge())
125                .userCity(userProfile.getCity())
126                .membershipLevel(userProfile.getMembershipLevel())
127                .interestTags(userProfile.getInterestTags())
128                .itemCategories(items.stream()
129                        .map(ItemDTO::getCategory)
130                        .distinct()
131                        .collect(Collectors.toList()))
132                .hourOfDay(LocalDateTime.now().getHour())
133                .dayOfWeek(LocalDateTime.now().getDayOfWeek().getValue())
134                .build();
135    }
136
137    /**
138     * 记录曝光事件（异步）
139     */
140    private void recordExposureEvent(Long userId, String scenario, List<RecommendedItem> items) {
141        List<Long> itemIds = items.stream()
142                .map(RecommendedItem::getItemId)
143                .collect(Collectors.toList());
144
145        ExposureEvent event = ExposureEvent.builder()
146                .userId(userId)
147                .scenario(scenario)
148                .itemIds(itemIds)
149                .timestamp(System.currentTimeMillis())
150                .build();
151
152        kafkaTemplate.send("recommendation-exposure", event);
153        log.debug("Exposure event sent for user: {}", userId);
154    }
155
156    /**
157     * 记录点击事件
158     */
159    public void recordClickEvent(Long userId, Long itemId, String scenario) {
160        ClickEvent event = ClickEvent.builder()
161                .userId(userId)
162                .itemId(itemId)
163                .scenario(scenario)
164                .timestamp(System.currentTimeMillis())
165                .build();
166
167        kafkaTemplate.send("recommendation-click", event);
168        log.info("Click event recorded for user: {}, item: {}", userId, itemId);
169    }
170}
171