Spring Cloud 分布式智能推荐系统实战

架构亮点

1. 特征与模型解耦：独立微服务交互，并行迭代互不影响。
2. 水平扩展能力：各服务独立扩容，应对大促流量洪峰。
3. 容错与降级：Resilience4j 实现熔断，模型不可用时降级为规则推荐。
4. 灰度发布支持：网关流量路由验证新算法效果。

AI 模型选型与训练

推荐算法选择需综合考虑场景、数据规模及实时性。电商和内容平台常见算法包括协同过滤、矩阵分解及深度学习模型。

算法选型分析

1. 协同过滤：经典算法，分基于用户和物品两种。实现简单可解释性强，但存在冷启动和稀疏性问题。
2. 矩阵分解：将评分矩阵分解为低维因子，捕捉潜在关系。SVD 及其变体是代表，工业界应用广泛。
3. 深度学习模型：如 NCF、DeepFM、Wide&Deep。能捕捉高阶特征交互，效果通常优于传统方法。

模型训练与导出

实际项目中常用 Python 生态的 PyTorch 或 TensorFlow 进行训练。以下是一个使用 PyTorch 实现 NCF 模型的简化示例，注意代码结构要清晰，导入库要规范：

# train_ncf_model.py
import torch
import torch.nn as nn
import numpy as np
from torch.utils.data import Dataset, DataLoader

# 定义 NCF 模型
class NCFModel(nn.Module):
    def __init__(self, num_users, num_items, embedding_dim=32):
        super(NCFModel, self).__init__()
        # 用户嵌入层
        self.user_embedding = nn.Embedding(num_users, embedding_dim)
        # 物品嵌入层
        self.item_embedding = nn.Embedding(num_items, embedding_dim)
        # MLP 层
        self.mlp = nn.Sequential(
            nn.Linear(embedding_dim * 2, 128),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(128, 64),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(64, 1),
            nn.Sigmoid()
        )

    def forward(self, user_ids, item_ids):
        user_emb = self.user_embedding(user_ids)
        item_emb = self.item_embedding(item_ids)
        concat = torch.cat([user_emb, item_emb], dim=-1)
        return self.mlp(concat)

# 自定义数据集
class RecommendationDataset(Dataset):
    def __init__(self, user_ids, item_ids, labels):
        self.user_ids = torch.LongTensor(user_ids)
        self.item_ids = torch.LongTensor(item_ids)
        self.labels = torch.FloatTensor(labels)

    def __len__(self):
        return len(self.labels)

    def __getitem__(self, idx):
        return self.user_ids[idx], self.item_ids[idx], self.labels[idx]

# 训练函数
def train_model(train_data, val_data, num_users, num_items, epochs=10):
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = NCFModel(num_users, num_items).to(device)
    criterion = nn.BCELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    train_loader = DataLoader(train_data, batch_size=256, shuffle=True)

    for epoch in range(epochs):
        model.train()
        total_loss = 0
        for user_ids, item_ids, labels in train_loader:
            user_ids, item_ids, labels = user_ids.to(device), item_ids.to(device), labels.to(device)
            optimizer.zero_grad()
            predictions = model(user_ids, item_ids).squeeze()
            loss = criterion(predictions, labels)
            loss.backward()
            optimizer.step()
            total_loss += loss.item()
        print(f'Epoch {epoch+1}/{epochs}, Loss: {total_loss/len(train_loader):.4f}')
    return model

# 导出模型为 ONNX 格式
def export_to_onnx(model, output_path='ncf_model.onnx'):
    model.eval()
    dummy_user_ids = torch.LongTensor([0])
    dummy_item_ids = torch.LongTensor([0])
    torch.onnx.export(
        model, (dummy_user_ids, dummy_item_ids), output_path,
        input_names=['user_ids', 'item_ids'],
        output_names=['prediction'],
        dynamic_axes={'user_ids': {0: 'batch_size'}, 'item_ids': {0: 'batch_size'}, 'prediction': {0: 'batch_size'}},
        opset_version=14
    )
    print(f'Model exported to {output_path}')

if __name__ == '__main__':
    # 模拟训练数据
    num_users = 10000
    num_items = 50000
    num_samples = 100000
    user_ids = np.random.randint(0, num_users, num_samples)
    item_ids = np.random.randint(0, num_items, num_samples)
    labels = np.random.randint(0, 2, num_samples).astype(float)

    # 划分训练集和验证集
    split_idx = int(0.8 * num_samples)
    train_data = RecommendationDataset(user_ids[:split_idx], item_ids[:split_idx], labels[:split_idx])
    val_data = RecommendationDataset(user_ids[split_idx:], item_ids[split_idx:], labels[split_idx:])

    # 训练模型
    model = train_model(train_data, val_data, num_users, num_items, epochs=5)
    # 导出为 ONNX 格式
    export_to_onnx(model)

模型服务化

训练完成后，部署为服务供 Java 应用调用。这里使用 FastAPI 实现模型推理服务，提供 REST API：

# model_server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import onnxruntime as ort
import numpy as np
from typing import List

app = FastAPI(title='Recommendation Model API')

# 加载 ONNX 模型
session = ort.InferenceSession('ncf_model.onnx')

class PredictionRequest(BaseModel):
    user_ids: List[int]
    item_ids: List[int]

class PredictionResponse(BaseModel):
    predictions: List[float]

@app.post('/api/v1/predict', response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # 准备输入数据
        user_ids = np.array(request.user_ids, dtype=np.int64).reshape(-1, 1)
        item_ids = np.array(request.item_ids, dtype=np.int64).reshape(-1, 1)
        # ONNX 推理
        inputs = {'user_ids': user_ids, 'item_ids': item_ids}
        predictions = session.run(None, inputs)[0]
        return PredictionResponse(predictions=predictions.tolist())
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get('/health')
async def health_check():
    return {'status': 'healthy'}

if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)

对于高性能场景，也可以考虑使用 gRPC 协议，比 REST 更高效。

核心微服务实现

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

项目结构与依赖

创建一个 Maven 多模块项目，统一管理版本：

<!-- pom.xml -->
<project>
    <groupId>com.example.recommendation</groupId>
    <artifactId>recommendation-system</artifactId>
    <version>1.0.0</version>
    <packaging>pom</packaging>
    <modules>
        <module>user-service</module>
        <module>item-service</module>
        <module>recommendation-service</module>
        <module>feature-engine</module>
        <module>common</module>
    </modules>
    <properties>
        <java.version>17</java.version>
        <spring-boot.version>3.2.0</spring-boot.version>
        <spring-cloud.version>2023.0.0</spring-cloud.version>
    </properties>
    <dependencyManagement>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
            <dependency>
                <groupId>org.springframework.cloud</groupId>
                <artifactId>spring-cloud-dependencies</artifactId>
                <version>${spring-cloud.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>
        </dependencies>
    </dependencyManagement>
</project>

User Service 实现

用户服务提供基础信息和画像查询接口。注意包声明和类结构的规范性：

// user-service/src/main/java/com/example/user/controller/UserController.java
package com.example.user.controller;

import com.example.user.dto.UserProfileDTO;
import com.example.user.service.UserService;
import lombok.RequiredArgsConstructor;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/users")
@RequiredArgsConstructor
public class UserController {
    private final UserService userService;

    /**
     * 获取用户画像
     */
    @GetMapping("/{userId}/profile")
    public UserProfileDTO getUserProfile(@PathVariable Long userId) {
        return userService.getUserProfile(userId);
    }

    /**
     * 批量获取用户画像
     */
    @PostMapping("/profiles/batch")
    public Map<Long, UserProfileDTO> getUserProfilesBatch(@RequestBody List<Long> userIds) {
        return userService.getUserProfilesBatch(userIds);
    }
}

// user-service/src/main/java/com/example/user/service/UserService.java
package com.example.user.service;

import com.example.user.dto.UserProfileDTO;
import com.example.user.entity.UserProfile;
import com.example.user.repository.UserProfileRepository;
import lombok.RequiredArgsConstructor;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
public class UserService {
    private final UserProfileRepository userProfileRepository;

    /**
     * 获取用户画像，使用 Redis 缓存
     */
    @Cacheable(value = "userProfiles", key = "#userId")
    public UserProfileDTO getUserProfile(Long userId) {
        UserProfile profile = userProfileRepository.findById(userId).orElseThrow(() -> new RuntimeException("User not found: " + userId));
        return UserProfileDTO.builder()
                .userId(profile.getUserId())
                .gender(profile.getGender())
                .age(profile.getAge())
                .city(profile.getCity())
                .membershipLevel(profile.getMembershipLevel())
                .interestTags(profile.getInterestTags())
                .build();
    }

    public Map<Long, UserProfileDTO> getUserProfilesBatch(List<Long> userIds) {
        List<UserProfile> profiles = userProfileRepository.findAllById(userIds);
        return profiles.stream().collect(Collectors.toMap(UserProfile::getUserId, profile -> UserProfileDTO.builder()
                .userId(profile.getUserId()).gender(profile.getGender()).age(profile.getAge())
                .city(profile.getCity()).membershipLevel(profile.getMembershipLevel())
                .interestTags(profile.getInterestTags()).build()));
    }
}

配置文件 application.yml 需正确配置 Nacos 和 Redis：

# user-service/src/main/resources/application.yml
server:
  port: 8081
spring:
  application:
    name: user-service
  datasource:
    url: jdbc:mysql://localhost:3306/user_db
    username: root
    password: password
  data:
    redis:
      host: localhost
      port: 6379
  cloud:
    nacos:
      discovery:
        server-addr: localhost:8848
        namespace: public
      config:
        server-addr: localhost:8848
        file-extension: yml

Recommendation Service 实现

推荐服务是核心，负责协调流程。首先定义 Feign 客户端，并配置降级工厂以增强鲁棒性：

// recommendation-service/src/main/java/com/example/recommendation/client/UserClient.java
package com.example.recommendation.client;

import com.example.common.dto.UserProfileDTO;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;

import java.util.List;
import java.util.Map;

@FeignClient(name = "user-service", path = "/api/users", fallbackFactory = UserClientFallback.class)
public interface UserClient {
    @GetMapping("/{userId}/profile")
    UserProfileDTO getUserProfile(@PathVariable("userId") Long userId);

    @PostMapping("/profiles/batch")
    Map<Long, UserProfileDTO> getUserProfilesBatch(@RequestBody List<Long> userIds);
}

降级工厂实现：当用户服务不可用时，返回默认画像，避免请求失败。

// recommendation-service/src/main/java/com/example/recommendation/client/UserClientFallback.java
package com.example.recommendation.client;

import com.example.common.dto.UserProfileDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class UserClientFallback implements UserClient {
    @Override
    public UserProfileDTO getUserProfile(Long userId) {
        log.warn("User service fallback triggered for userId: {}", userId);
        return UserProfileDTO.builder()
                .userId(userId).gender("unknown").age(25).city("unknown")
                .membershipLevel("NORMAL").interestTags(Collections.emptyList()).build();
    }

    @Override
    public Map<Long, UserProfileDTO> getUserProfilesBatch(List<Long> userIds) {
        log.warn("User service batch fallback triggered");
        return Collections.emptyMap();
    }
}

推荐服务核心逻辑：

// recommendation-service/src/main/java/com/example/recommendation/service/RecommendationService.java
package com.example.recommendation.service;

import com.example.common.dto.*;
import com.example.recommendation.client.ItemClient;
import com.example.recommendation.client.UserClient;
import com.example.recommendation.client.ModelClient;
import com.example.recommendation.config.RecommendationProperties;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor
public class RecommendationService {
    private final UserClient userClient;
    private final ItemClient itemClient;
    private final ModelClient modelClient;
    private final FeatureEngineClient featureEngineClient;
    private final RedisTemplate<String, Object> redisTemplate;
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private final RecommendationProperties properties;

    /**
     * 获取个性化推荐
     */
    public RecommendationResult getRecommendations(Long userId, String scenario, int size) {
        // 1. 尝试从缓存获取
        String cacheKey = String.format("recommendation:%d:%s", userId, scenario);
        List<RecommendedItem> cachedResult = (List<RecommendedItem>) redisTemplate.opsForValue().get(cacheKey);
        if (cachedResult != null) {
            log.info("Cache hit for user: {}, scenario: {}", userId, scenario);
            return RecommendationResult.builder().userId(userId).scenario(scenario).items(cachedResult).source("CACHE").build();
        }

        // 2. 获取用户画像
        UserProfileDTO userProfile = userClient.getUserProfile(userId);

        // 3. 获取候选商品集
        List<Long> candidateItemIds = getCandidateItems(userProfile, scenario, size * 10);

        // 4. 批量获取商品信息
        List<ItemDTO> candidateItems = itemClient.getItemsBatch(candidateItemIds);

        // 5. 特征工程
        FeatureVector featureVector = buildFeatureVector(userId, userProfile, candidateItems);

        // 6. 模型推理评分
        Map<Long, Double> itemScores = modelClient.predict(userId, candidateItemIds, featureVector);

        // 7. 排序并返回 Top-K
        List<RecommendedItem> recommendedItems = candidateItems.stream()
                .filter(item -> itemScores.containsKey(item.getItemId()))
                .sorted((a, b) -> Double.compare(itemScores.get(b.getItemId()), itemScores.get(a.getItemId())))
                .limit(size)
                .map(item -> RecommendedItem.builder()
                        .itemId(item.getItemId())
                        .itemName(item.getItemName())
                        .category(item.getCategory())
                        .price(item.getPrice())
                        .score(itemScores.get(item.getItemId()))
                        .reason("基于您的兴趣推荐")
                        .build())
                .collect(Collectors.toList());

        // 8. 缓存结果
        redisTemplate.opsForValue().set(cacheKey, recommendedItems, Duration.ofMinutes(properties.getCacheExpireMinutes()));

        // 9. 异步记录曝光事件
        recordExposureEvent(userId, scenario, recommendedItems);

        return RecommendationResult.builder().userId(userId).scenario(scenario).items(recommendedItems).source("MODEL").build();
    }

    private List<Long> getCandidateItems(UserProfileDTO userProfile, String scenario, int poolSize) {
        List<String> interests = userProfile.getInterestTags();
        if (interests.isEmpty()) {
            interests = Arrays.asList("热门");
        }
        return itemClient.getItemsByCategory(interests.get(0)).stream()
                .limit(poolSize)
                .map(ItemDTO::getItemId)
                .collect(Collectors.toList());
    }

    private FeatureVector buildFeatureVector(Long userId, UserProfileDTO userProfile, List<ItemDTO> items) {
        return FeatureVector.builder()
                .userId(userId)
                .userGender(userProfile.getGender())
                .userAge(userProfile.getAge())
                .userCity(userProfile.getCity())
                .membershipLevel(userProfile.getMembershipLevel())
                .interestTags(userProfile.getInterestTags())
                .itemCategories(items.stream().map(ItemDTO::getCategory).distinct().collect(Collectors.toList()))
                .hourOfDay(LocalDateTime.now().getHour())
                .dayOfWeek(LocalDateTime.now().getDayOfWeek().getValue())
                .build();
    }

    private void recordExposureEvent(Long userId, String scenario, List<RecommendedItem> items) {
        List<Long> itemIds = items.stream().map(RecommendedItem::getItemId).collect(Collectors.toList());
        ExposureEvent event = ExposureEvent.builder()
                .userId(userId).scenario(scenario).itemIds(itemIds)
                .timestamp(System.currentTimeMillis()).build();
        kafkaTemplate.send("recommendation-exposure", event);
        log.debug("Exposure event sent for user: {}", userId);
    }

    public void recordClickEvent(Long userId, Long itemId, String scenario) {
        ClickEvent event = ClickEvent.builder()
                .userId(userId).itemId(itemId).scenario(scenario)
                .timestamp(System.currentTimeMillis()).build();
        kafkaTemplate.send("recommendation-click", event);
        log.info("Click event recorded for user: {}, item: {}", userId, itemId);
    }
}

模型推理客户端通过 WebClient 调用 Python 模型服务，增加了超时控制和重试机制：

// recommendation-service/src/main/java/com/example/recommendation/client/ModelClient.java
package com.example.recommendation.client;

import com.example.recommendation.dto.ModelPredictRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class ModelClient {
    private final WebClient.Builder webClientBuilder;

    @Value("${model.service.url}")
    private String modelServiceUrl;

    @Value("${model.service.timeout}")
    private int timeoutMs;

    public Map<Long, Double> predict(Long userId, List<Long> itemIds, FeatureVector featureVector) {
        WebClient webClient = webClientBuilder.baseUrl(modelServiceUrl).build();
        List<Integer> userIds = Collections.nCopies(itemIds.size(), userId.intValue());
        ModelPredictRequest request = ModelPredictRequest.builder()
                .userIds(userIds)
                .itemIds(itemIds.stream().map(Long::intValue).collect(Collectors.toList()))
                .featureVector(featureVector)
                .build();

        Map<Long, Double> scores = new HashMap<>();
        try {
            ModelPredictResponse response = webClient.post()
                    .uri("/api/v1/predict")
                    .bodyValue(request)
                    .retrieve()
                    .bodyToMono(ModelPredictResponse.class)
                    .timeout(Duration.ofMillis(timeoutMs))
                    .retryWhen(Retry.backoff(3, Duration.ofMillis(100)))
                    .block();
            if (response != null && response.getPredictions() != null) {
                for (int i = 0; i < itemIds.size(); i++) {
                    scores.put(itemIds.get(i), response.getPredictions().get(i));
                }
            }
        } catch (Exception e) {
            log.error("Model prediction failed for userId: {}, error: {}", userId, e.getMessage());
            itemIds.forEach(id -> scores.put(id, 0.5));
        }
        return scores;
    }
}

熔断与降级配置

使用 Resilience4j 实现服务熔断和降级策略，确保核心业务不受影响：

// recommendation-service/src/main/java/com/example/recommendation/config/ResilienceConfig.java
package com.example.recommendation.config;

import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.timelimiter.TimeLimiterConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class ResilienceConfig {
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
                .slidingWindowSize(10)
                .minimumNumberOfCalls(5)
                .failureRateThreshold(50)
                .waitDurationInOpenState(Duration.ofSeconds(10))
                .permittedNumberOfCallsInHalfOpenState(3)
                .slowCallDurationThreshold(Duration.ofSeconds(3))
                .slowCallRateThreshold(50)
                .build();
        return CircuitBreakerRegistry.of(config);
    }

    @Bean
    public TimeLimiterConfig timeLimiterConfig() {
        return TimeLimiterConfig.custom().timeoutDuration(Duration.ofSeconds(5)).build();
    }
}

降级策略实现：当模型服务不可用时，返回热门商品推荐作为兜底。

// recommendation-service/src/main/java/com/example/recommendation/fallback/RecommendationFallback.java
package com.example.recommendation.fallback;

import com.example.common.dto.RecommendationResult;
import com.example.common.dto.RecommendedItem;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;

@Slf4j
@Component
public class RecommendationFallback {
    public RecommendationResult getHotItemsFallback(Long userId, String scenario) {
        log.warn("Model service degraded, using hot items fallback for user: {}", userId);
        List<RecommendedItem> hotItems = getHotItems(scenario);
        return RecommendationResult.builder().userId(userId).scenario(scenario).items(hotItems).source("FALLBACK_HOT_ITEMS").build();
    }

    private List<RecommendedItem> getHotItems(String scenario) {
        return Arrays.asList(
                RecommendedItem.builder().itemId(1001L).itemName("热门商品 1").category("电子").price(299.0).score(0.95).reason("热门推荐").build(),
                RecommendedItem.builder().itemId(1002L).itemName("热门商品 2").category("服装").price(199.0).score(0.92).reason("热门推荐").build()
        );
    }
}

异步日志收集

通过 Kafka 发送用户行为事件，用于在线学习和模型更新：

// recommendation-service/src/main/java/com/example/recommendation/producer/EventProducer.java
package com.example.recommendation.producer;

import com.example.common.dto.ExposureEvent;
import com.example.common.dto.ClickEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Slf4j
@Component
@RequiredArgsConstructor
public class EventProducer {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public void sendExposureEvent(ExposureEvent event) {
        kafkaTemplate.send("recommendation-exposure", event).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.debug("Exposure event sent successfully: {}", event);
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("Failed to send exposure event: {}", event, ex);
            }
        });
    }

    public void sendClickEvent(ClickEvent event) {
        kafkaTemplate.send("recommendation-click", event).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("Click event sent successfully: {}", event);
            }

            @Override
            public void onFailure(Throwable ex) {
                log.error("Failed to send click event: {}", event, ex);
            }
        });
    }
}

部署与性能优化

开发完成后，需要考虑生产环境部署和性能优化。

Docker 容器化

为每个服务创建 Dockerfile，例如 recommendation-service：

# recommendation-service/Dockerfile
FROM eclipse-temurin:17-jre-alpine
WORKDIR /app
COPY target/recommendation-service-*.jar app.jar
ENV JAVA_OPTS="-Xms512m -Xmx1g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"
EXPOSE 8083
ENTRYPOINT ["sh", "-c", "java $JAVA_OPTS -jar app.jar"]

使用 Docker Compose 编排本地开发环境，包含 Nacos、MySQL、Redis、Kafka 等服务。

Kubernetes 编排

生产环境推荐使用 Kubernetes。以下是 recommendation-service 的部署配置示例，包含资源限制和探针：

# k8s/recommendation-service-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: recommendation-service
  labels:
    app: recommendation-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: recommendation-service
  template:
    metadata:
      labels:
        app: recommendation-service
    spec:
      containers:
      - name: recommendation-service
        image: your-registry/recommendation-service:1.0.0
        ports:
        - containerPort: 8083
        env:
        - name: SPRING_PROFILES_ACTIVE
          value: "prod"
        - name: SPRING_CLOUD_NACOS_DISCOVERY_SERVER_ADDR
          value: "nacos-service:8848"
        - name: SPRING_KAFKA_BOOTSTRAP_SERVERS
          value: "kafka-service:9092"
        - name: MODEL_SERVICE_URL
          value: "http://model-service:8000"
        - name: JAVA_OPTS
          value: "-Xms1g -Xmx2g -XX:+UseG1GC -XX:MaxGCPauseMillis=200"
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "2000m"
        livenessProbe:
          httpGet:
            path: /actuator/health
            port: 8083
          initialDelaySeconds: 60
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /actuator/health
            port: 8083
          initialDelaySeconds: 30
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: recommendation-service
spec:
  selector:
    app: recommendation-service
  ports:
  - protocol: TCP
    port: 8083
    targetPort: 8083
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: recommendation-service-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: recommendation-service
  minReplicas: 3
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

性能优化策略

1. 模型服务 GPU 加速

深度学习模型推理建议使用 GPU。Dockerfile 基于 NVIDIA CUDA 镜像，并在 K8s 中申请 GPU 资源：

resources:
  limits:
    nvidia.com/gpu: 1

2. 多级缓存策略

实现三级缓存减少计算压力：

// CacheConfiguration.java
@Configuration
@EnableCaching
public class CacheConfiguration {
    @Bean
    public RedisCacheManager redisCacheManager(RedisConnectionFactory factory) {
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofMinutes(30))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
        return RedisCacheManager.builder(factory).cacheDefaults(config).withInitialCacheConfigurations(getCacheConfigurations()).build();
    }

    private Map<String, RedisCacheConfiguration> getCacheConfigurations() {
        Map<String, RedisCacheConfiguration> configMap = new HashMap<>();
        // 用户画像缓存 - 1 小时
        configMap.put("userProfiles", RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofHours(1)));
        // 推荐结果缓存 - 30 分钟
        configMap.put("recommendations", RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMinutes(30)));
        // 热门商品缓存 - 10 分钟
        configMap.put("hotItems", RedisCacheConfiguration.defaultCacheConfig().entryTtl(Duration.ofMinutes(10)));
        return configMap;
    }
}

3. JVM 调优

针对推荐服务内存和 CPU 特性调整参数：

JAVA_OPTS="-Xms2g -Xmx4g -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:G1ReservePercent=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/logs/heapdump.hprof -XX:+PrintGCDetails -XX:+PrintGCDateStamps -Xloggc:/logs/gc.log -Duser.timezone=Asia/Shanghai -Dfile.encoding=UTF-8"

4. 数据库优化

• 高频查询字段添加索引。
• 读写分离分担主库压力。
• 热点数据使用 Redis 缓存。

-- 用户画像表索引优化
CREATE INDEX idx_user_id ON user_profile(user_id);
CREATE INDEX idx_membership ON user_profile(membership_level);
CREATE INDEX idx_city ON user_profile(city);

-- 商品表索引优化
CREATE INDEX idx_item_category ON item(category);
CREATE INDEX idx_item_brand ON item(brand);
CREATE INDEX idx_item_created ON item(created_at);

压测结果

使用 JMeter 进行压力测试，测试场景为并发获取推荐结果。

测试环境：

• 推荐服务：3 个实例，每实例 2C4G
• 模型服务：2 个实例，每实例 4C8G + GPU
• Redis：1 主 2 从
• Kafka：3 节点集群

测试结果对比：

优化后结果（启用多级缓存 + JVM 调优）：

通过缓存和 JVM 调优，系统性能提升了约 100-150%，P99 延迟降低了约 50%。

总结与展望

本文详细介绍了基于 Spring Cloud 微服务架构构建智能推荐系统的完整过程，涵盖架构设计、模型选型、服务实现到部署优化，展示了如何在企业级 Java 生态中集成 AI 能力。通过微服务架构，我们实现了特征工程与模型推理的解耦，使各组件可以独立开发、部署和扩展；通过服务熔断和降级策略，确保了系统的高可用性；通过多级缓存和 JVM 调优，显著提升了系统性能。

Spring Cloud 与 AI 的结合为传统 Java 应用注入了智能化能力，开发者可以在熟悉的 Java 生态中轻松集成机器学习模型。这种架构的优势在于技术栈灵活性、系统可扩展性及部署便捷性。

展望未来，推荐系统还有许多值得探索的方向：

1. 实时推荐：引入 Flink 等流处理技术，实现基于用户实时行为的毫秒级推荐更新。
2. 多模态特征融合：处理文本、图像、视频等多种模态特征，实现跨模态推荐。
3. 联邦学习与隐私保护：在用户设备本地进行模型训练，保护用户隐私。
4. AutoML 与模型自动迭代：实现模型的自动训练、评估和部署，形成闭环迭代体系。
5. 大模型增强推荐：利用 LLM 的理解能力，实现更精准的用户意图识别和推荐理由生成。

AI 技术的快速发展为推荐系统带来了新的机遇和挑战。作为 Java 开发者，我们需要保持开放心态，积极拥抱新技术，在保证系统稳定性和可维护性的前提下，将 AI 能力优雅地集成到现有架构中。