SpringBoot实战：如何用Milvus 2.2.3打造高性能推荐系统（附完整代码）

最近在重构一个内容推荐项目时，我面临一个核心挑战：如何在海量的用户和内容向量数据中，实现毫秒级的相似度匹配。传统的基于标签或协同过滤的方法，在应对复杂的非结构化数据（如图像、文本的语义特征）时，显得力不从心。这时，向量数据库进入了我的视野。经过一番技术选型，我最终锁定了Milvus，特别是其2.x版本，它在性能和功能上都有了质的飞跃。这篇文章，我想和你分享的，不仅仅是如何在SpringBoot项目里引入一个依赖、写几行配置，而是如何从零开始，构建一个真正能扛住线上流量、具备业务灵活性的推荐系统核心引擎。我们会深入Milvus 2.2.3的过滤筛选、连接管理、查询优化等实战细节，并附上经过生产环境验证的完整代码。

1. 项目架构与Milvus 2.x核心优势

在推荐系统的技术栈中，向量检索层正变得越来越关键。简单来说，我们的目标是将用户的历史行为（如点击、浏览、收藏）和内容（商品、文章、视频）都通过深度学习模型转化为高维向量（即Embedding）。当用户发起请求时，系统需要快速找到与用户向量最“相似”的Top-K个内容向量。这个过程，就是向量相似度搜索。

为什么是Milvus？在评估了多个开源向量数据库后，我发现Milvus 2.x版本在架构上做了重大革新。它采用了存储与计算分离的云原生设计，这使得扩展性变得极其灵活。但对我们Java开发者而言，最直观的吸引力来自以下几点：

  • 强大的过滤能力：这是从1.x升级到2.x最诱人的理由。在1.x时代，先检索再过滤是常见做法，效率低下。2.x版本支持在向量检索的同时，进行标量字段的过滤（比如 status = 1 AND category in (‘tech‘, ‘news’)），这能直接将无效结果排除在计算之外，性能提升显著。
  • 完善的SDK与生态：Milvus提供了功能齐全的Java SDK，与SpringBoot的集成路径清晰。社区活跃，遇到问题比较容易找到解决方案。
  • 生产级可靠性：支持数据持久化、高可用部署，并且具备监控告警能力，能满足线上系统的稳定性要求。

在我们的推荐场景中，内容向量通常是离线由Spark或Flink任务批量生成并灌入Milvus。而SpringBoot服务作为在线查询层，需要以极低的延迟响应C端用户的请求。因此，我们设计的架构重点在于在线服务的稳定性、查询的极致性能以及应对复杂业务逻辑的灵活性。

注意：虽然向量生成是离线过程，但确保在线查询服务与向量数据版本的一致性，是架构设计中需要仔细考虑的一环，通常通过版本号或时间分区来管理。

2. 工程搭建：从Maven依赖到健壮配置

万事开头难，一个清晰的工程结构能避免后续很多麻烦。我们从一个全新的SpringBoot 2.7+项目开始。

2.1 依赖管理与版本选择

首先，在pom.xml中引入核心依赖。这里我强烈建议直接使用2.2.3或更高版本，避免使用已停止维护的1.x版本。

<dependency>
    <groupId>io.milvus</groupId>
    <artifactId>milvus-sdk-java</artifactId>
    <version>2.2.3</version>
</dependency>

为了更好的依赖管理和避免冲突，可以锁定Milvus SDK相关组件的版本。同时，确保你的项目中有合理的日志框架（如Logback）和JSON处理库（Jackson），因为SDK内部会用到。

2.2 配置注入与客户端工厂模式

将Milvus的连接信息放在application.yml中是标准做法，但直接写在业务代码里会缺乏灵活性。我更喜欢用一个独立的配置类来集中管理。

# application.yml
milvus:
  client:
    host: ${MILVUS_HOST:127.0.0.1} # 支持环境变量覆盖
    port: 19530
    token: # 如果启用认证则填写
    database-name: default # 2.x支持多数据库
    connect-timeout-ms: 10000
    keep-alive-time-ms: 30000

接下来，创建配置类MilvusConfig。这里我采用了工厂模式来封装客户端的创建，好处是便于后续扩展，例如支持客户端池化、不同环境的配置切换等。

@Configuration
@ConfigurationProperties(prefix = "milvus.client")
@Data // 使用Lombok简化代码
public class MilvusConfig {
    private String host;
    private Integer port;
    private String token;
    private String databaseName;
    private Long connectTimeoutMs;
    private Long keepAliveTimeMs;

    @Bean(destroyMethod = "close")
    public MilvusServiceClient milvusServiceClient() {
        ConnectParam connectParam = ConnectParam.newBuilder()
                .withHost(host)
                .withPort(port)
                .withAuthorization(token) // 可选
                .withDatabaseName(databaseName) // 2.x新特性
                .withConnectTimeout(connectTimeoutMs.intValue(), TimeUnit.MILLISECONDS)
                .withKeepAliveTimeout(keepAliveTimeMs.intValue(), TimeUnit.MILLISECONDS)
                .build();
        return new MilvusServiceClient(connectParam);
    }
}

提示：在生产环境中，务必配置合理的超时时间。connectTimeout影响建立连接的等待时间，keepAliveTimeout则用于维护长连接，防止因网络波动导致的不必要重连。

2.3 客户端生命周期与异常处理

将MilvusServiceClient声明为Spring Bean并由容器管理其生命周期是最佳实践。destroyMethod = “close”确保了应用关闭时，客户端连接能被正确释放，避免资源泄漏。

此外，我们需要一个全局的异常处理策略。Milvus SDK的操作可能会抛出MilvusException或网络超时异常。我建议定义一个自定义的业务异常，并在Service层进行统一的捕获和转换。

@Service
@Slf4j
public class VectorSearchService {
    @Autowired
    private MilvusServiceClient client;

    public List<Long> searchWithFallback(String collectionName, List<List<Float>> vector, int topK) {
        try {
            // ... 构建查询参数
            R<SearchResults> resp = client.search(searchParam);
            if (resp.getStatus() != R.Status.Success.getCode()) {
                log.error("Milvus search failed, code: {}, reason: {}", resp.getStatus(), resp.getMessage());
                // 这里可以触发降级策略，例如返回缓存的热门结果
                return Collections.emptyList();
            }
            return extractIds(resp.getData());
        } catch (MilvusException e) {
            log.error("Milvus service error during search", e);
            throw new BusinessException("向量检索服务暂时不可用", e);
        } catch (Exception e) {
            log.error("Unexpected error during vector search", e);
            throw new BusinessException("系统内部错误", e);
        }
    }
}

3. 核心查询：同步、异步与过滤实战

配置好客户端，我们就进入了最核心的部分：查询。Milvus 2.2.3的查询功能相当丰富，我们将拆解几个最常用的场景。

3.1 基础同步查询

这是最直接的查询方式，发送请求后阻塞等待结果。适用于对实时性要求高的核心推荐接口。

public List<Long> basicSearch(String collectionName,
                               List<List<Float>> queryVectors,
                               String vectorFieldName,
                               int topK,
                               MetricType metricType) {
    // 1. 参数校验
    if (CollectionUtils.isEmpty(queryVectors)) {
        return Collections.emptyList();
    }
    // 2. 构建搜索参数
    String searchParams = "{\"nprobe\": 16}"; // nprobe是IVF类索引的关键参数，影响精度和速度
    SearchParam searchParam = SearchParam.newBuilder()
            .withCollectionName(collectionName)
            .withMetricType(metricType) // 常用 IP(内积) 或 L2(欧氏距离)
            .withVectorFieldName(vectorFieldName)
            .withVectors(queryVectors)
            .withTopK(topK)
            .withParams(searchParams)
            .build();
    // 3. 执行查询
    R<SearchResults> response = milvusClient.search(searchParam);
    // 4. 结果提取 (假设主键是Int类型)
    return response.getData()
                   .getResults()
                   .getIds()
                   .getIntId()
                   .getDataList();
}

关键参数解析：

  • nprobe: 在索引类型为IVF_FLAT或IVF_SQ8时使用。它决定了搜索时遍历的聚类中心数量。值越大，精度越高，但耗时越长。需要在精度和性能之间权衡，通常通过离线测试确定一个最优值。
  • MetricType: 相似度度量方式。IP（内积）常用于余弦相似度（向量需提前归一化），L2（欧氏距离）则更为通用。

3.2 威力倍增：带过滤条件的查询

这是2.x版本的精髓。假设我们的内容向量集合中，每个向量都附带了一些标量属性，比如category（分类）、status（状态）、publish_time（发布时间）。我们希望在检索相似内容时，只返回status为上架状态且category属于用户偏好的内容。

public List<Long> filteredSearch(String collectionName,
                                  List<List<Float>> queryVector,
                                  int topK,
                                  List<String> preferredCategories) {
    // 构建过滤表达式
    // 表达式语法类似SQL的WHERE子句，支持 AND, OR, >, <, ==, in 等
    String expr = "status == 1 and category in [\"" +
                  String.join("\", \"", preferredCategories) +
                  "\"]";

    SearchParam searchParam = SearchParam.newBuilder()
            .withCollectionName(collectionName)
            .withMetricType(MetricType.IP)
            .withVectorFieldName("embedding")
            .withVectors(queryVector)
            .withTopK(topK)
            .withExpr(expr) // 关键：传入过滤表达式
            .withParams("{\"nprobe\": 16}")
            .build();

    R<SearchResults> response = milvusClient.search(searchParam);
    return extractIds(response.getData());
}

这个功能将原本需要在应用层做的“先查后滤”两步操作，合并为数据库层的一步高效操作，极大地减少了网络传输和数据处理的开销。

3.3 提升吞吐：异步查询与Future

对于某些实时性要求稍低、但吞吐量要求高的场景，例如后台数据分析、用户行为日志的批量相似性计算，可以使用异步查询。

public CompletableFuture<List<Long>> asyncSearch(String collectionName,
                                                   List<List<Float>> queryVectors,
                                                   int topK) {
    SearchParam searchParam = SearchParam.newBuilder()
            .withCollectionName(collectionName)
            .withMetricType(MetricType.L2)
            .withVectorFieldName("vector")
            .withVectors(queryVectors)
            .withTopK(topK)
            .withParams("{\"nprobe\": 8}") // 异步任务可适当降低精度求速度
            .build();

    // 提交异步任务，返回ListenableFuture
    ListenableFuture<R<SearchResults>> future = milvusClient.searchAsync(searchParam);

    // 转换为更现代的CompletableFuture，便于组合异步操作
    CompletableFuture<List<Long>> resultFuture = new CompletableFuture<>();
    Futures.addCallback(future, new FutureCallback<R<SearchResults>>() {
        @Override
        public void onSuccess(R<SearchResults> result) {
            resultFuture.complete(extractIds(result.getData()));
        }
        @Override
        public void onFailure(Throwable t) {
            resultFuture.completeExceptionally(t);
        }
    }, MoreExecutors.directExecutor()); // 使用直接执行器，回调在当前线程执行

    return resultFuture;
}

使用异步查询时，你可以利用CompletableFuture进行复杂的异步编排，比如同时发起多个不相关的向量查询，然后等待所有结果返回，最大化利用I/O等待时间。

4. 高级特性与性能优化

掌握了基本查询后，要打造高性能系统，还需要关注一些高级特性和优化点。

4.1 分区管理提升查询效率

当单个集合的数据量非常庞大时（例如数亿条），即使有索引，查询效率也可能下降。Milvus支持分区概念，你可以按照某种逻辑（如时间：2023Q1, 2023Q2；或业务类型：article, video）将数据划分到不同分区。查询时可以指定分区，从而大幅缩小搜索范围。

// 假设我们按内容类型分区
public List<Long> searchWithinPartition(String collectionName,
                                         List<List<Float>> queryVector,
                                         int topK,
                                         String partitionTag) {
    List<String> partitionNames = Collections.singletonList(partitionTag);

    SearchParam searchParam = SearchParam.newBuilder()
            .withCollectionName(collectionName)
            .withPartitionNames(partitionNames) // 指定分区
            .withMetricType(MetricType.IP)
            .withVectorFieldName("embedding")
            .withVectors(queryVector)
            .withTopK(topK)
            .withParams("{\"nprobe\": 10}")
            .build();

    R<SearchResults> response = milvusClient.search(searchParam);
    return extractIds(response.getData());
}

如何选择分区键？这需要根据你的查询模式来决定。如果你的查询总是附带content_type=‘video‘这样的过滤条件，那么按content_type分区就是很好的选择。

4.2 索引创建策略与参数调优

向量检索的速度和精度，极大程度上依赖于索引。Milvus支持多种索引类型，适用于不同场景。

索引类型适用场景特点内存占用
FLAT小规模数据集（<100万），追求100%精确率暴力计算，无需训练，精度最高高
IVF_FLAT中等规模数据集，平衡精度与速度基于聚类的倒排索引，需训练，nprobe可调中
IVF_SQ8大规模数据集，追求高压缩比和速度在IVF_FLAT基础上进行标量化压缩，有精度损失低
HNSW各种规模，追求高召回率基于图算法，无需训练，查询速度快，建索引慢高

对于推荐系统，IVF系列索引是常见选择。创建索引的代码通常在线下执行：

public void createIndex(String collectionName, String fieldName) {
    final IndexType INDEX_TYPE = IndexType.IVF_FLAT;
    final String INDEX_PARAM = "{\"nlist\":1024}"; // nlist是聚类中心数，通常设为 sqrt(数据量)

    MilvusClient client = ... // 获取客户端
    CreateIndexParam createIndexReq = CreateIndexParam.newBuilder()
            .withCollectionName(collectionName)
            .withFieldName(fieldName)
            .withIndexType(INDEX_TYPE)
            .withMetricType(MetricType.IP)
            .withExtraParam(INDEX_PARAM)
            .build();

    client.createIndex(createIndexReq);
}

调优建议：

  1. nlist：创建索引时指定。对于千万级数据，设置为 4096 或 8192 是合理的起点。
  2. nprobe：查询时指定。nprobe / nlist 的比例决定了搜索的广度。通常从 nlist 的 1/16 到 1/4 开始测试。比例越大，越慢但越准。
  3. 批量测试：使用你的真实数据集，固定topK，遍历不同的(nlist, nprobe)组合，绘制精度-耗时曲线，找到业务可接受的平衡点。

4.3 连接池与资源管理

在高并发场景下，为每次查询创建新的连接是灾难性的。虽然Milvus Java SDK的MilvusServiceClient内部有一定连接复用机制，但在SpringBoot中，我们仍需注意：

  • 单例Bean：确保MilvusServiceClient在整个应用中是单例的，如前文配置所示。
  • 监控与健康检查：集成Spring Boot Actuator，自定义一个健康指示器，定期执行一个简单的hasCollection操作来检查Milvus连接状态。
  • 超时与重试：为SDK配置合理的超时，并结合Spring Retry或Resilience4j为关键的搜索操作添加重试逻辑，应对网络抖动。
@Bean
public MilvusServiceClient milvusClient(MilvusConfig config) {
    ConnectParam connectParam = ... // 构建参数
    // 可以在这里配置更底层的Netty通道参数，影响连接池行为
    return new MilvusServiceClient(connectParam);
}

5. 完整实战：一个推荐API的诞生

让我们把上面的知识点串联起来，实现一个完整的、带缓存的推荐API端点。

第一步：定义服务层

@Service
@Slf4j
public class RecommendationService {
    @Autowired
    private MilvusServiceClient milvusClient;
    @Autowired
    private CacheManager cacheManager; // 假设使用Spring Cache

    private static final String COLLECTION_NAME = "content_vectors";
    private static final String VECTOR_FIELD = "embedding";

    @Cacheable(value = "userRecommendation", key = "#userId", unless = "#result.empty")
    public List<ContentDTO> recommendForUser(Long userId, int topK) {
        // 1. 获取用户向量 (从用户特征服务或缓存)
        List<Float> userVector = getUserVectorFromFeatureStore(userId);
        if (userVector == null) {
            return getFallbackHotContents(); // 降级策略
        }

        // 2. 获取用户偏好标签
        Set<String> userPreferences = getUserPreferences(userId);

        // 3. 构建Milvus过滤表达式
        String expr = null;
        if (!CollectionUtils.isEmpty(userPreferences)) {
            expr = String.format("status == 1 and category in [\"%s\"]",
                    String.join("\", \"", userPreferences));
        } else {
            expr = "status == 1"; // 默认只返回上架内容
        }

        // 4. 执行向量搜索
        List<List<Float>> searchVectors = Collections.singletonList(userVector);
        SearchParam param = SearchParam.newBuilder()
                .withCollectionName(COLLECTION_NAME)
                .withMetricType(MetricType.IP)
                .withVectorFieldName(VECTOR_FIELD)
                .withVectors(searchVectors)
                .withTopK(topK * 2) // 多查一些，供后续业务过滤
                .withExpr(expr)
                .withParams("{\"nprobe\": 20}")
                .build();

        R<SearchResults> resp = milvusClient.search(param);
        List<Long> contentIds = extractIds(resp.getData());

        // 5. 获取内容详情并排序 (可能结合CTR模型进行精排)
        List<ContentDTO> contents = getContentDetails(contentIds);
        return rerankByBusinessRule(contents).stream().limit(topK).collect(Collectors.toList());
    }

    private List<Long> extractIds(SearchResults results) {
        // ... 省略具体实现，处理IntId或StringId
        return results.getResults().getIds().getIntId().getDataList();
    }
    // ... 其他辅助方法
}

第二步：暴露REST API

@RestController
@RequestMapping("/api/v1/recommend")
@Validated
public class RecommendationController {
    @Autowired
    private RecommendationService recommendationService;

    @GetMapping("/for-you")
    public ApiResponse<List<ContentDTO>> getForYouRecommendation(
            @RequestParam @Min(1) @Max(50) Integer size,
            HttpServletRequest request) {
        Long userId = getCurrentUserId(request); // 从Token或Session获取用户ID
        List<ContentDTO> recommendations = recommendationService.recommendForUser(userId, size);
        return ApiResponse.success(recommendations);
    }
}

这个流程体现了线上推荐系统的典型模式：召回（Milvus向量检索）-> 粗排/过滤（业务规则）-> 精排（复杂模型）-> 返回。Milvus在其中承担了高效“召回”的核心职责。

在整个集成和优化过程中，我最大的体会是，理解业务查询模式比盲目调参更重要。是先过滤再向量搜索，还是先搜索再过滤？数据是否需要分区？索引类型和参数如何选择？这些问题都需要结合你的实际数据量、并发请求和业务逻辑来回答。多花时间在测试环境用真实数据做基准测试，记录下不同配置下的QPS、延迟和召回率，这些数据将是你在生产环境做出正确决策的最有力依据。

Logo
乐奇 Rokid 开放社区

更多推荐

cover

一句话打造「丢哪里」垃圾分类AIUI智能体

avatar 乐奇 Rokid 开放社区
cover

从源码拆解一个 AIUI 每日一词智能体：页面、离线索引与硬件交互

avatar 乐奇 Rokid 开放社区
cover

Rokid Glasses 2 的三层补能系统：263mAh 眼镜电池、胶囊电池与充电盒，如何转化为可量产方案？

avatar 乐奇 Rokid 开放社区