546 lines
18 KiB
C#
546 lines
18 KiB
C#
using System.Diagnostics;
|
|
using MAX.Core.DependencyInjection;
|
|
using MAX.Framework.Commons;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.Logging;
|
|
using OpenSearch.Client;
|
|
using OpenSearch.Net;
|
|
using Store.Modules.CloudProduct.Dtos;
|
|
using Store.Modules.CloudProduct.Entities;
|
|
using Store.Modules.CloudProduct.Enums;
|
|
|
|
namespace Store.Modules.CloudProduct.Services
|
|
{
|
|
public interface IOpenSearchService : ISingletonDependency
|
|
{
|
|
Task AddDocumentAsync(GoodsSearchDocument document);
|
|
Task BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents);
|
|
Task<bool> ClearIndexAsync();
|
|
Task<long> CountDocumentsAsync();
|
|
Task<bool> CreateIndexAsync();
|
|
Task DeleteDocumentsAsync(IEnumerable<long> ids);
|
|
Task DeleteIndexAsync();
|
|
Task<OpenSearchStatusDto> GetStatusAsync();
|
|
Task<bool> IndexExistsAsync();
|
|
Task<bool> IsIndexMappingHealthyAsync();
|
|
Task RefreshIndexAsync();
|
|
Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList);
|
|
Task<(long Total, List<GoodsSearchDocument> Documents)> SearchByTitleAsync(GoodsSearchRequest request, GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance);
|
|
}
|
|
|
|
internal class OpenSearchService : IOpenSearchService
|
|
{
|
|
private readonly string _nodeUrl;
|
|
private readonly string _indexName;
|
|
private readonly IOpenSearchClient _client = null!;
|
|
|
|
// 查询类操作（search/count/status/ping）统一超时
|
|
private static readonly TimeSpan QueryTimeout = TimeSpan.FromSeconds(5);
|
|
// 批量写入/建索引/删除索引等管理类操作超时（数据量大时需要更长，可挪到配置文件）
|
|
private static readonly TimeSpan AdminTimeout = TimeSpan.FromSeconds(30);
|
|
private readonly ILogger<OpenSearchService> _logger;
|
|
|
|
public OpenSearchService(IConfiguration configuration,ILogger<OpenSearchService> logger)
|
|
{
|
|
var config = configuration.GetSection("Search");
|
|
_nodeUrl = config.GetValue<string>("Url") ?? "";
|
|
_indexName = config.GetValue<string>("IndexName") ?? "";
|
|
_client = CreateClient();
|
|
_logger = logger;
|
|
}
|
|
|
|
private IOpenSearchClient CreateClient()
|
|
{
|
|
var pool = new SingleNodeConnectionPool(new Uri(_nodeUrl));
|
|
var settings = new ConnectionSettings(pool)
|
|
.DefaultIndex(_indexName)
|
|
.DefaultMappingFor<GoodsSearchDocument>(m => m.IndexName(_indexName))
|
|
.RequestTimeout(QueryTimeout)
|
|
.MaxRetryTimeout(QueryTimeout)
|
|
.PrettyJson()
|
|
.DisableDirectStreaming(); // 便于调试打印请求/响应报文
|
|
|
|
return new OpenSearchClient(settings);
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 1. 创建索引 ============
|
|
/// </summary>
|
|
public async Task<bool> CreateIndexAsync()
|
|
{
|
|
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (existsResponse.Exists)
|
|
{
|
|
if (await IsIndexMappingHealthyAsync())
|
|
{
|
|
Console.WriteLine($"索引 [{_indexName}] 已存在且映射正确，无需重复创建。");
|
|
return true;
|
|
}
|
|
|
|
Console.WriteLine($"索引 [{_indexName}] 的映射不完整（可能由清空索引等操作造成），正在重建...");
|
|
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!deleteResponse.IsValid)
|
|
{
|
|
throw new AppException("删除损坏索引失败: " + deleteResponse.DebugInformation);
|
|
}
|
|
}
|
|
|
|
// 优先尝试使用 IK 中文分词（需集群安装 analysis-ik 插件）
|
|
var createResponse = await TryCreateIndexAsync(useIk: true);
|
|
|
|
if (!createResponse.IsValid)
|
|
{
|
|
Console.WriteLine("IK 分词插件不可用，已回退到 standard 分析器，中文分词将失效，请检查集群是否安装 analysis-ik 插件。");
|
|
createResponse = await TryCreateIndexAsync(useIk: false);
|
|
}
|
|
|
|
if (createResponse.IsValid)
|
|
{
|
|
Console.WriteLine($"索引 [{_indexName}] 创建成功。");
|
|
return true;
|
|
}
|
|
else
|
|
{
|
|
throw new AppException("索引创建失败: " + createResponse.DebugInformation);
|
|
}
|
|
}
|
|
|
|
public async Task<bool> IsIndexMappingHealthyAsync()
|
|
{
|
|
var mappingStatus = await GetIndexMappingStatusAsync();
|
|
return mappingStatus.Healthy;
|
|
}
|
|
|
|
private async Task<(bool Healthy, string? IndexAnalyzer, string? SearchAnalyzer)> GetIndexMappingStatusAsync()
|
|
{
|
|
var getResponse = await _client.Indices.GetAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!getResponse.IsValid)
|
|
{
|
|
throw new AppException("读取索引映射失败: " + getResponse.DebugInformation);
|
|
}
|
|
if (!getResponse.Indices.TryGetValue(_indexName, out var indexState))
|
|
{
|
|
return (false, null, null);
|
|
}
|
|
|
|
var mapping = indexState.Mappings;
|
|
if (mapping?.Properties == null
|
|
|| !mapping.Properties.TryGetValue(new PropertyName("title"), out var property)
|
|
|| property is not ITextProperty textProperty)
|
|
{
|
|
return (false, null, null);
|
|
}
|
|
|
|
var indexAnalyzer = textProperty.Analyzer;
|
|
var searchAnalyzer = textProperty.SearchAnalyzer;
|
|
var usesIk = indexAnalyzer == "ik_max_word" && searchAnalyzer == "ik_smart";
|
|
var usesStandard = indexAnalyzer == "standard" && searchAnalyzer == "standard";
|
|
return (usesIk || usesStandard, indexAnalyzer, searchAnalyzer);
|
|
}
|
|
|
|
private async Task<CreateIndexResponse> TryCreateIndexAsync(bool useIk)
|
|
{
|
|
var indexAnalyzer = useIk ? "ik_max_word" : "standard";
|
|
var searchAnalyzer = useIk ? "ik_smart" : "standard";
|
|
|
|
return await _client.Indices.CreateAsync(_indexName, c => c
|
|
.Settings(s => s
|
|
.NumberOfShards(1)
|
|
.NumberOfReplicas(0)
|
|
)
|
|
.Map<GoodsSearchDocument>(m => m
|
|
.Properties(p => p
|
|
.Number(n => n.Name(x => x.Id).Type(NumberType.Long))
|
|
.Text(t => t
|
|
.Name(x => x.Title)
|
|
.Analyzer(indexAnalyzer)
|
|
.SearchAnalyzer(searchAnalyzer)
|
|
)
|
|
.Number(n => n.Name(x => x.CatId1).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId2).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId3).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.CatId4).Type(NumberType.Integer))
|
|
.Number(n => n.Name(x => x.MinPrice).Type(NumberType.Long))
|
|
.Number(n => n.Name(x => x.SoldQuantity).Type(NumberType.Integer))
|
|
)
|
|
)
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
|
|
);
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 2. 删除索引 ============
|
|
/// </summary>
|
|
public async Task DeleteIndexAsync()
|
|
{
|
|
var response = await _client.Indices.DeleteAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!response.IsValid)
|
|
{
|
|
throw new AppException("删除失败: " + response.DebugInformation);
|
|
}
|
|
}
|
|
|
|
public async Task<bool> IndexExistsAsync()
|
|
{
|
|
var response = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
return response.Exists;
|
|
}
|
|
|
|
public async Task RefreshIndexAsync()
|
|
{
|
|
var response = await _client.Indices.RefreshAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!response.IsValid)
|
|
{
|
|
throw new AppException("刷新索引失败: " + response.DebugInformation);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 3.1 批量写入商品搜索文档（存在则覆盖） ============
|
|
/// </summary>
|
|
public async Task BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents)
|
|
{
|
|
if (documents == null) throw new ArgumentNullException(nameof(documents));
|
|
|
|
var list = documents.ToList();
|
|
if (list.Count == 0) return;
|
|
|
|
// 确保索引已按 IK 分词映射创建，否则中文分词无效
|
|
await CreateIndexAsync();
|
|
|
|
var bulkResponse = await _client.BulkAsync(b => b
|
|
.Index(_indexName)
|
|
.IndexMany(list, (descriptor, doc) => descriptor.Id(doc.Id))
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout))
|
|
);
|
|
|
|
if (!bulkResponse.IsValid)
|
|
{
|
|
throw new AppException("写入 OpenSearch 失败: " + bulkResponse.DebugInformation);
|
|
}
|
|
|
|
// 刷新索引，保证数据立即可搜索（仅用于演示，生产环境不建议频繁 Refresh）
|
|
await RefreshIndexAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 3.2 根据数据库商品重建搜索索引 ============
|
|
/// </summary>
|
|
public async Task<long> RebuildIndexAsync(IEnumerable<GoodsEntity> goodsList)
|
|
{
|
|
if (goodsList == null) throw new ArgumentNullException(nameof(goodsList));
|
|
|
|
var documents = goodsList
|
|
.Where(x => x != null && x.GoodsId > 0)
|
|
.Select(x => new GoodsSearchDocument
|
|
{
|
|
Id = x.GoodsId,
|
|
Title = x.Title ?? string.Empty,
|
|
CatId1 = x.CatId1,
|
|
CatId2 = x.CatId2,
|
|
CatId3 = x.CatId3,
|
|
CatId4 = x.CatId4,
|
|
MinPrice = x.MinPrice,
|
|
SoldQuantity = x.SoldQuantity
|
|
})
|
|
.ToList();
|
|
|
|
if (documents.Count == 0) return 0;
|
|
|
|
await BulkIndexGoodsAsync(documents);
|
|
return documents.Count;
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 3.3 按商品编号批量删除 OpenSearch 文档（软/硬删除共用） ============
|
|
/// </summary>
|
|
public async Task DeleteDocumentsAsync(IEnumerable<long> ids)
|
|
{
|
|
var list = ids?.Distinct().ToList() ?? new List<long>();
|
|
if (list.Count == 0) return;
|
|
|
|
var exists = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!exists.Exists)
|
|
{
|
|
return;
|
|
}
|
|
|
|
var response = await _client.BulkAsync(b =>
|
|
{
|
|
foreach (var id in list)
|
|
{
|
|
b.Delete<GoodsSearchDocument>(d => d.Index(_indexName).Id(id));
|
|
}
|
|
b.RequestConfiguration(r => r.RequestTimeout(AdminTimeout));
|
|
return b;
|
|
});
|
|
|
|
if (!response.IsValid)
|
|
{
|
|
throw new AppException("删除 OpenSearch 文档失败: " + response.DebugInformation);
|
|
}
|
|
|
|
await RefreshIndexAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 4. 手动新增一条搜索文档 ============
|
|
/// </summary>
|
|
public async Task AddDocumentAsync(GoodsSearchDocument document)
|
|
{
|
|
if (document == null) throw new ArgumentNullException(nameof(document));
|
|
|
|
var response = await _client.IndexAsync(document, i => i
|
|
.Index(_indexName)
|
|
.Id(document.Id)
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
|
);
|
|
|
|
if (response.IsValid)
|
|
{
|
|
await RefreshIndexAsync();
|
|
}
|
|
else
|
|
{
|
|
throw new AppException("新增失败: " + response.DebugInformation);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 5. 分词搜索（分页 + 类目/价格/销量/平台过滤 + 排序） ============
|
|
/// </summary>
|
|
/// <param name="request">搜索条件（关键词、类目、价格、销量、平台、分页）</param>
|
|
/// <param name="sortBy">排序方式</param>
|
|
public async Task<(long Total, List<GoodsSearchDocument> Documents)> SearchByTitleAsync(GoodsSearchRequest request,
|
|
GoodsSearchSortBy sortBy = GoodsSearchSortBy.Relevance)
|
|
{
|
|
var keyword = request.Title;
|
|
int? catId1 = request.CatId1;
|
|
int? catId2 = request.CatId2;
|
|
int? catId3 = request.CatId3;
|
|
int? catId4 = request.CatId4;
|
|
long? minPrice = request.MinPrice;
|
|
long? maxPrice = request.MaxPrice;
|
|
int? minSoldQuantity = request.MinSoldQuantity;
|
|
int? maxSoldQuantity = request.MaxSoldQuantity;
|
|
int? platform = request.Platform;
|
|
int pageIndex = request.PageIndex;
|
|
int pageSize = request.PageSize;
|
|
|
|
if (pageIndex < 1) pageIndex = 1;
|
|
if (pageSize < 1) pageSize = 20;
|
|
if (pageSize > 2000) pageSize = 2000; // 避免深分页/单页过大拖垮性能，按需调整上限
|
|
|
|
var from = (pageIndex - 1) * pageSize;
|
|
|
|
ISearchResponse<GoodsSearchDocument> searchResponse;
|
|
try
|
|
{
|
|
searchResponse = await _client.SearchAsync<GoodsSearchDocument>(s => s
|
|
.Index(_indexName)
|
|
.Query(q => q
|
|
.Bool(b =>
|
|
{
|
|
if (string.IsNullOrWhiteSpace(keyword) == false)
|
|
{
|
|
b.Must(m => m.Match(mm => mm.Field(f => f.Title).Query(keyword)));
|
|
}
|
|
|
|
var filters = new List<Func<QueryContainerDescriptor<GoodsSearchDocument>, QueryContainer>>();
|
|
if (platform.HasValue && platform!.Value > 0) filters.Add(f => f.Term(t => t.Field(x => x.Platform).Value(platform.Value)));
|
|
|
|
if (catId1.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId1).Value(catId1.Value)));
|
|
if (catId2.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId2).Value(catId2.Value)));
|
|
if (catId3.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId3).Value(catId3.Value)));
|
|
if (catId4.HasValue) filters.Add(f => f.Term(t => t.Field(x => x.CatId4).Value(catId4.Value)));
|
|
|
|
if (minPrice.HasValue || maxPrice.HasValue)
|
|
{
|
|
filters.Add(f => f.Range(r =>
|
|
{
|
|
var range = r.Field(x => x.MinPrice);
|
|
if (minPrice.HasValue) range = range.GreaterThanOrEquals(minPrice.Value);
|
|
if (maxPrice.HasValue) range = range.LessThanOrEquals(maxPrice.Value);
|
|
return range;
|
|
}));
|
|
}
|
|
|
|
if (minSoldQuantity.HasValue || maxSoldQuantity.HasValue)
|
|
{
|
|
filters.Add(f => f.Range(r =>
|
|
{
|
|
var range = r.Field(x => x.SoldQuantity);
|
|
if (minSoldQuantity.HasValue) range = range.GreaterThanOrEquals(minSoldQuantity.Value);
|
|
if (maxSoldQuantity.HasValue) range = range.LessThanOrEquals(maxSoldQuantity.Value);
|
|
return range;
|
|
}));
|
|
}
|
|
|
|
if (filters.Count > 0) b.Filter(filters.ToArray());
|
|
return b;
|
|
})
|
|
)
|
|
.Sort(sort => sortBy switch
|
|
{
|
|
GoodsSearchSortBy.PriceAsc => sort.Ascending(f => f.MinPrice),
|
|
GoodsSearchSortBy.PriceDesc => sort.Descending(f => f.MinPrice),
|
|
GoodsSearchSortBy.SoldDesc => sort.Descending(f => f.SoldQuantity),
|
|
_ => sort.Descending(SortSpecialField.Score)
|
|
})
|
|
.From(from)
|
|
.Size(pageSize)
|
|
.TrackTotalHits(true) // 命中数超过 10000 时默认不精确统计，需要显式开启
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
|
);
|
|
}
|
|
catch (OpenSearchClientException ex) when (ex.Response?.Success == false)
|
|
{
|
|
throw new AppException($"搜索超时或失败（{QueryTimeout.TotalSeconds}秒）: " + ex.Message);
|
|
}
|
|
|
|
if (!searchResponse.IsValid)
|
|
{
|
|
throw new AppException("搜索失败: " + searchResponse.DebugInformation);
|
|
}
|
|
|
|
return (searchResponse.Total, searchResponse.Documents.ToList());
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 6. 统计文档数 ============
|
|
/// </summary>
|
|
public async Task<long> CountDocumentsAsync()
|
|
{
|
|
var countResponse = await _client.CountAsync<GoodsSearchDocument>(c => c
|
|
.Index(_indexName)
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout))
|
|
);
|
|
if (countResponse.IsValid)
|
|
{
|
|
Console.WriteLine($"索引 [{_indexName}] 当前文档总数: {countResponse.Count}");
|
|
return countResponse.Count;
|
|
}
|
|
else
|
|
{
|
|
Console.WriteLine("统计失败: " + countResponse.DebugInformation);
|
|
throw new AppException(countResponse.DebugInformation);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// ============ 7. 查询 OpenSearch 状态 ============
|
|
/// </summary>
|
|
public async Task<OpenSearchStatusDto> GetStatusAsync()
|
|
{
|
|
var stopwatch = Stopwatch.StartNew();
|
|
var status = new OpenSearchStatusDto
|
|
{
|
|
NodeUrl = _nodeUrl,
|
|
IndexName = _indexName,
|
|
};
|
|
|
|
try
|
|
{
|
|
var ping = await _client.PingAsync(p => p
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
status.Connected = ping.IsValid;
|
|
if (!ping.IsValid)
|
|
{
|
|
status.Message = ping.DebugInformation;
|
|
_logger.LogError(ping.DebugInformation);
|
|
return status;
|
|
}
|
|
_logger.LogInformation("OpenSerach 已连接");
|
|
var clusterHealth = await _client.Cluster.HealthAsync(Indices.All, h => h
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!clusterHealth.IsValid)
|
|
{
|
|
status.Message = clusterHealth.DebugInformation;
|
|
return status;
|
|
}
|
|
|
|
status.ClusterName = clusterHealth.ClusterName;
|
|
status.ClusterStatus = clusterHealth.Status.ToString().ToLowerInvariant();
|
|
status.NumberOfNodes = clusterHealth.NumberOfNodes;
|
|
status.NumberOfDataNodes = clusterHealth.NumberOfDataNodes;
|
|
status.ActivePrimaryShards = clusterHealth.ActivePrimaryShards;
|
|
status.ActiveShards = clusterHealth.ActiveShards;
|
|
status.RelocatingShards = clusterHealth.RelocatingShards;
|
|
status.InitializingShards = clusterHealth.InitializingShards;
|
|
status.UnassignedShards = clusterHealth.UnassignedShards;
|
|
status.IndexExists = await IndexExistsAsync();
|
|
|
|
if (status.IndexExists)
|
|
{
|
|
var mappingStatus = await GetIndexMappingStatusAsync();
|
|
status.MappingHealthy = mappingStatus.Healthy;
|
|
status.IndexAnalyzer = mappingStatus.IndexAnalyzer;
|
|
status.SearchAnalyzer = mappingStatus.SearchAnalyzer;
|
|
|
|
var count = await _client.CountAsync<GoodsSearchDocument>(c => c
|
|
.Index(_indexName)
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
status.DocumentCount = count.IsValid ? count.Count : -1;
|
|
if (!count.IsValid)
|
|
{
|
|
status.Message = count.DebugInformation;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
status.Message = $"索引 [{_indexName}] 不存在";
|
|
}
|
|
|
|
status.Healthy = status.IndexExists
|
|
&& status.MappingHealthy
|
|
&& status.ClusterStatus != "red"
|
|
&& status.DocumentCount >= 0;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
status.Healthy = false;
|
|
status.Message = ex.Message;
|
|
}
|
|
finally
|
|
{
|
|
stopwatch.Stop();
|
|
status.ElapsedMilliseconds = stopwatch.ElapsedMilliseconds;
|
|
}
|
|
|
|
return status;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 清空索引：删除后按统一映射重建，重建期间该索引不可查（由后台定时任务负责回填数据，接受短暂空窗）
|
|
/// </summary>
|
|
public async Task<bool> ClearIndexAsync()
|
|
{
|
|
var existsResponse = await _client.Indices.ExistsAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(QueryTimeout)));
|
|
if (!existsResponse.Exists)
|
|
{
|
|
Console.WriteLine($"索引 {_indexName} 不存在");
|
|
return false;
|
|
}
|
|
|
|
var deleteResponse = await _client.Indices.DeleteAsync(_indexName, d => d
|
|
.RequestConfiguration(r => r.RequestTimeout(AdminTimeout)));
|
|
if (!deleteResponse.IsValid)
|
|
{
|
|
Console.WriteLine($"删除索引 {_indexName} 失败: {deleteResponse.DebugInformation}");
|
|
return false;
|
|
}
|
|
Console.WriteLine($"索引 {_indexName} 已删除");
|
|
|
|
return await CreateIndexAsync();
|
|
}
|
|
}
|
|
}
|