fix: 修复批量更新bug,修改为bulkApi的方式进行批量更新
update: 获取时间轴数据,更新轨迹内容为空的轨迹
This commit is contained in:
parent
c11615775b
commit
faa7b6cd7e
@ -298,7 +298,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
|||||||
track = existTracks.FirstOrDefault();
|
track = existTracks.FirstOrDefault();
|
||||||
|
|
||||||
if (track == null) continue; // 不存在不让改
|
if (track == null) continue; // 不存在不让改
|
||||||
|
|
||||||
}
|
}
|
||||||
else
|
else
|
||||||
{
|
{
|
||||||
@ -454,12 +453,8 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
|
|||||||
|
|
||||||
if (updateTracks.Any())
|
if (updateTracks.Any())
|
||||||
{
|
{
|
||||||
//逐条更新
|
//批量差异化更新轨迹
|
||||||
foreach (var track in updateTracks)
|
await _trackDetailsEsRepository.BulkUpdateAsync(updateTracks);
|
||||||
{
|
|
||||||
await _trackDetailsEsRepository.UpdateAsync(track);
|
|
||||||
}
|
|
||||||
await _trackDetailsEsRepository.UpdateManyAsync(updateTracks);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return insertTracks;
|
return insertTracks;
|
||||||
|
|||||||
@ -195,6 +195,8 @@ public class TrackOrderService : ApplicationService
|
|||||||
return new UnifyResultDto<List<GetDetailTimelineOutput>>(false, "未找到指定id的子系统");
|
return new UnifyResultDto<List<GetDetailTimelineOutput>>(false, "未找到指定id的子系统");
|
||||||
|
|
||||||
var trackDetails = await _trackDetailsEsRepository.GetListAsync(order.Id);
|
var trackDetails = await _trackDetailsEsRepository.GetListAsync(order.Id);
|
||||||
|
//时间轴不显示轨迹内容为空的轨迹,进行过滤
|
||||||
|
trackDetails = trackDetails.Where(x => !x.TrackContent.IsNullOrWhiteSpace() && x.TrackContent != GlobalConst.EmptyString).ToList();
|
||||||
res.Data = new List<GetDetailTimelineOutput>();
|
res.Data = new List<GetDetailTimelineOutput>();
|
||||||
|
|
||||||
foreach (var item in trackDetails)
|
foreach (var item in trackDetails)
|
||||||
|
|||||||
@ -18,6 +18,8 @@ public interface ITrackDetailsEsRepository : IElasticsearchClientService<TrackDe
|
|||||||
|
|
||||||
Task UpdateManyAsync(List<TrackDetailsIndex> indexes, int maxRetries = 3);
|
Task UpdateManyAsync(List<TrackDetailsIndex> indexes, int maxRetries = 3);
|
||||||
|
|
||||||
|
Task BulkUpdateAsync(List<TrackDetailsIndex> indexes, int maxRetries = 3);
|
||||||
|
|
||||||
Task AddManyAsync(List<TrackDetailsIndex> index);
|
Task AddManyAsync(List<TrackDetailsIndex> index);
|
||||||
|
|
||||||
Task AddAsync(TrackDetailsIndex track);
|
Task AddAsync(TrackDetailsIndex track);
|
||||||
|
|||||||
@ -265,6 +265,86 @@ public class TrackDetailsEsRepository : ElasticSearchContextAbstract<TrackDetail
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public async Task BulkUpdateAsync(List<TrackDetailsIndex> indexes, int maxRetries = 3)
|
||||||
|
{
|
||||||
|
if (indexes.Count == 0) return;
|
||||||
|
|
||||||
|
//构建批量更新操作描述符
|
||||||
|
var desc = new BulkDescriptor(GlobalConst.TrackDetailsIndex);
|
||||||
|
foreach (var track in indexes)
|
||||||
|
{
|
||||||
|
desc = desc.AddOperation(new BulkUpdateDescriptor<TrackDetailsIndex, object>()
|
||||||
|
.Id(track.Id)
|
||||||
|
.Index(track.IndexName)
|
||||||
|
.Script(s => s.Source(@"
|
||||||
|
ctx._source.trackPlace = params.trackPlace;
|
||||||
|
ctx._source.trackTime = params.trackTime;
|
||||||
|
ctx._source.trackTimeZone = params.trackTimeZone;
|
||||||
|
ctx._source.updateUser = params.updateUser;
|
||||||
|
ctx._source.updateTime = params.updateTime;
|
||||||
|
ctx._source.trackContent = params.trackContent;
|
||||||
|
ctx._source.trackEnContent = params.trackEnContent;
|
||||||
|
ctx._source.isVisibility = params.isVisibility;
|
||||||
|
ctx._source.remarks = params.remarks;
|
||||||
|
ctx._source.files = params.files;
|
||||||
|
ctx._source.trackPlaceCity = params.trackPlaceCity;
|
||||||
|
ctx._source.trackPlaceCityId = params.trackPlaceCityId;
|
||||||
|
ctx._source.trackPlaceProvince = params.trackPlaceProvince;
|
||||||
|
ctx._source.trackPlaceProvinceId = params.trackPlaceProvinceId;
|
||||||
|
ctx._source.trackPlaceNation = params.trackPlaceNation;
|
||||||
|
ctx._source.trackPlaceNationId = params.trackPlaceNationId;
|
||||||
|
ctx._source.trackPlacePostcode = params.trackPlacePostcode;
|
||||||
|
ctx._source.destinationTimeZone = params.destinationTimeZone;
|
||||||
|
ctx._source.estimatedDeliveryDate = params.estimatedDeliveryDate;
|
||||||
|
").Params(p => p
|
||||||
|
.Add("trackPlace", track.TrackPlace)
|
||||||
|
.Add("trackTime", track.TrackTime)
|
||||||
|
.Add("trackTimeZone", track.TrackTimeZone)
|
||||||
|
.Add("updateUser", track.UpdateUser)
|
||||||
|
.Add("updateTime", DateTime.Now)
|
||||||
|
//S06新增字段更新
|
||||||
|
.Add("trackContent", track.TrackContent)
|
||||||
|
.Add("trackEnContent", track.TrackEnContent)
|
||||||
|
.Add("isVisibility", track.IsVisibility)
|
||||||
|
.Add("remarks", track.Remarks)
|
||||||
|
.Add("files", track.Files)
|
||||||
|
.Add("trackPlaceCity", track.TrackPlaceCity)
|
||||||
|
.Add("trackPlaceCityId", track.TrackPlaceCityId)
|
||||||
|
.Add("trackPlaceProvince", track.TrackPlaceProvince)
|
||||||
|
.Add("trackPlaceProvinceId", track.TrackPlaceProvinceId)
|
||||||
|
.Add("trackPlaceNation", track.TrackPlaceNation)
|
||||||
|
.Add("trackPlaceNationId", track.TrackPlaceNationId)
|
||||||
|
.Add("trackPlacePostcode", track.TrackPlacePostcode)
|
||||||
|
.Add("destinationTimeZone", track.DestinationTimeZone)
|
||||||
|
.Add("estimatedDeliveryDate", track.EstimatedDeliveryDate))
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
var retryCount = 0;
|
||||||
|
var updateSuccessful = false;
|
||||||
|
while (retryCount < maxRetries && !updateSuccessful)
|
||||||
|
{
|
||||||
|
var response = await Context.BulkAsync(desc);
|
||||||
|
if (response.IsValid)
|
||||||
|
{
|
||||||
|
updateSuccessful = true;
|
||||||
|
}
|
||||||
|
else if (response.ServerError?.Status == 409) // 409 表示版本冲突
|
||||||
|
{
|
||||||
|
retryCount++;
|
||||||
|
if (retryCount >= maxRetries)
|
||||||
|
{
|
||||||
|
throw new InvalidOperationException(
|
||||||
|
$"轨迹更新冲突达到最大次数,track:{indexes.Count}条,原因:{response.DebugInformation}。");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
else
|
||||||
|
{
|
||||||
|
throw new InvalidOperationException($"轨迹更新失败,track:{indexes.Count}条,原因:{response.DebugInformation}。");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
|
public async Task AddManyAsync(List<TrackDetailsIndex> tracks)
|
||||||
{
|
{
|
||||||
// 按天写入es
|
// 按天写入es
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user