Files

569 lines
20 KiB
C#

using MAX.Plugins;
using Microsoft.Extensions.DependencyInjection;
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<long[]> 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;
private readonly IFreeSql _database;
private readonly GoodsIndexLock _indexLock;
public OpenSearchService([FromKeyedServices("CloudProduct")] PluginContext context, ILogger<OpenSearchService> logger,
[FromKeyedServices("CloudProductConnection")] IFreeSql database, GoodsIndexLock indexLock)
{
_database = database;
_indexLock = indexLock;
var config = context.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<long[]> BulkIndexGoodsAsync(IEnumerable<GoodsSearchDocument> documents)
{
if (documents == null) throw new ArgumentNullException(nameof(documents));
var list = documents.ToList();
if (list.Count == 0) return [];
await using var lease = await _indexLock.AcquireAsync();
var ids = list.Select(x => x.Id).Distinct().ToArray();
var blocked = (await _database.Select<GoodsViolationEntity>()
.Where(x => ids.Contains(x.GoodsId)).ToListAsync(x => x.GoodsId)).ToHashSet();
list = list.Where(x => !blocked.Contains(x.Id)).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 || bulkResponse.Errors)
{
throw new AppException("写入 OpenSearch 失败: " + bulkResponse.DebugInformation);
}
// 刷新索引,保证数据立即可搜索(仅用于演示,生产环境不建议频繁 Refresh)
await RefreshIndexAsync();
return list.Select(x => x.Id).Distinct().ToArray();
}
/// <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
{
Platform = x.Platform,
ThumbUrl = x.ThumbUrl,
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();
return await BulkIndexGoodsAsync(documents);
}
/// <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)
{
if (exists.ApiCall?.HttpStatusCode == 404) return;
throw new AppException("检查 OpenSearch 索引失败: " + exists.DebugInformation);
}
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.ApiCall?.HttpStatusCode is not (>= 200 and < 300)
|| response.Items.Count != list.Count
|| response.Items.Any(x => x.Status != 404 && (x.Error != null || x.Status < 200 || x.Status >= 300)))
{
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));
await using var lease = await _indexLock.AcquireAsync();
if (await _database.Select<GoodsViolationEntity>().AnyAsync(x => x.GoodsId == document.Id))
throw new AppException("违规商品不能写入搜索索引");
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()
{
await using var lease = await _indexLock.AcquireAsync();
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} 已删除");
var created = await CreateIndexAsync();
await _database.Update<GoodsEntity>().Set(x => x.IsSegmented, (byte)0)
.Where(x => x.Status == 0).ExecuteAffrowsAsync();
return created;
}
}
}