基于SpringBoot+Elasticsearch实现一套完整的搜索服务系统

/**
 * 搜索索引初始化：应用启动时创建索引与映射
 */
@Service
@RequiredArgsConstructor
public class SearchIndexInitializer {
    private final ElasticsearchClient es;
    private static final String INDEX = "zhiguang_content_index";

    @PostConstruct
    public void ensureIndex() {
        try {
            // 检查索引是否存在
            boolean exists = es.indices().exists(e -> e.index(INDEX)).value();
            if (exists) return;
            // 创建索引并定义映射
            es.indices().create(c -> c.index(INDEX).mappings(m -> m
                .properties("content_id", p -> p.long_(LongNumberProperty.of(b -> b)))
                .properties("title", p -> p.text(t -> t.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                .properties("body", p -> p.text(t -> t.analyzer("ik_max_word")))
                .properties("status", p -> p.keyword(KeywordProperty.of(b -> b)))
                .properties("title_suggest", p -> p.completion(CompletionProperty.of(b -> b)))
                // 其他字段省略...
            ));
        } catch (Exception ignored) {}
    }
}

@PostConstruct
public void ensureBackfill() {
    long cnt = es.count(c -> c.index(INDEX)).count();
    if (cnt > 0) return;
    int limit = 500;
    int offset = 0;
    while (true) {
        List<KnowPostFeedRow> rows = knowPostMapper.listFeedPublic(limit, offset);
        if (rows == null || rows.isEmpty()) break;
        for (KnowPostFeedRow r : rows) {
            upsertKnowPost(r.getId());
        }
        offset += rows.size();
    }
}

public void softDeleteKnowPost(long id) {
    Map<String, Object> doc = new HashMap<>();
    doc.put("content_id", id);
    doc.put("status", "deleted");
    es.index(i -> i.index(INDEX).id(String.valueOf(id))
        .document(doc).refresh(Refresh.WaitFor));
}

/**
 * 搜索索引 Outbox 消费者
 */
@Service
@RequiredArgsConstructor
public class CanalOutboxConsumerSearch {
    private final SearchIndexService indexService;

    @KafkaListener(topics = OutboxTopics.CANAL_OUTBOX, groupId = "search-index-consumer")
    public void onMessage(String message, Acknowledgment ack) {
        try {
            List<JsonNode> rows = OutboxMessageUtil.extractRows(objectMapper, message);
            for (JsonNode row : rows) {
                JsonNode payload = objectMapper.readTree(row.get("payload").asText());
                String entity = payload.get("entity").asText();
                String op = payload.get("op").asText();
                Long id = payload.get("id").asLong();
                if (!"knowpost".equals(entity) || id == null) continue;
                // 执行更新或软删除
                if ("delete".equalsIgnoreCase(op)) {
                    indexService.softDeleteKnowPost(id);
                } else {
                    indexService.upsertKnowPost(id);
                }
            }
            ack.acknowledge();
        } catch (Exception ignored) {}
    }
}

// 构建查询核心逻辑
.query(qb -> qb.functionScore(fs -> fs
    .query(qb2 -> qb2.bool(bq -> {
        // 多字段匹配，标题权重3倍
        bq.must(m -> m.multiMatch(mm -> mm.query(q).fields("title^3", "body")));
        // 过滤已发布内容
        bq.filter(f -> f.term(t -> t.field("status").value("published")));
        // 标签过滤
        if (!tags.isEmpty()) {
            bq.filter(f -> f.terms(t -> t.field("tags").terms(tv -> tv.value(tags))));
        }
        return bq;
    }))
    // 点赞数加权：log(1+like)×2
    .functions(fn -> fn.fieldValueFactor(f -> f.field("like_count").modifier(Log1p)).weight(2.0))
    // 浏览数加权：log(1+view)×1
    .functions(fn -> fn.fieldValueFactor(f -> f.field("view_count").modifier(Log1p)).weight(1.0))
    .boostMode(Sum)
))

public SuggestResponse suggest(String prefix, int size) {
    var resp = es.search(s -> s.index(INDEX)
        .suggest(sug -> sug.suggesters("title_suggest",
            sc -> sc.prefix(prefix).completion(c -> c.field("title_suggest").size(size))))
        , Map.class);
    // 解析建议结果并返回
    List<String> items = new ArrayList<>();
    resp.suggest().get("title_suggest").forEach(s -> {
        s.completion().options().forEach(opt -> items.add(opt.text()));
    });
    return new SuggestResponse(items);
}