ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update:新增/修改轨迹延时发送变更广播，目的为了等待es实际写盘

Browse Source
This commit is contained in:
hardstylewyl 2025-01-16 11:51:26 +08:00
parent 3ca803fe5b
commit e0be935489
6 changed files with 27 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/SJZY.Track.Application.Contract/Services/IBroadcastingService.cs
View File

@ -12,7 +12,7 @@ public interface IBroadcastingService : ITransientDependency
/// <param name="eto"></param>
/// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode);
void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode);
/// <summary>
/// 轨迹变更广播（删除）
@ -20,7 +20,7 @@ public interface IBroadcastingService : ITransientDependency
/// <param name="eto"></param>
/// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode);
void TrackChangePublicTopic(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode);
/// <summary>
/// 轨迹变更广播（删除）
@ -28,7 +28,7 @@ public interface IBroadcastingService : ITransientDependency
/// <param name="eto"></param>
/// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackChangeEto eto, string subSystemCode);
void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode);
/// <summary>
/// 轨迹推送完成广播（推送完成回调）

4
src/SJZY.Track.Application/CAP/EventHandlers/TrackDeleteEventHandler.cs
View File

@ -147,7 +147,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
var item = dataItem.First();
var content = contentList.First(m => m.TrackContentCode == item.TrackCode);
// _broadcastingService.TrackChangePublicTopicAsync(track,
// _broadcastingService.TrackChangePublicTopic(track,
// dataItem.Select(m => new OrderIdentifier
// {
// OrderId = m.OrderId,
@ -162,7 +162,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
//将每个订单的每个轨迹的删除变更发布
foreach (var track in dataItem)
{
_broadcastingService.TrackChangePublicTopicAsync(new TrackChangeEto
_broadcastingService.TrackChangePublicTopic(new TrackChangeEto
{
EstimatedDeliveryDate = track.EstimatedDeliveryDate,
TrackPlacePostcode = track.TrackPlacePostcode,

2
src/SJZY.Track.Application/CAP/EventHandlers/TrackPushEventHandler.cs
View File

@ -408,7 +408,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
if (config.IsAllPushWithLatest)
{
//同code同比较字段保证唯一
var old = body!.Data
var old = body.Data
.Where(x =>
{
//反序列化出的JsonElement和字典字符串类型做比较

25
src/SJZY.Track.Application/CAP/EventHandlers/TrackUpsertEventHandler.cs
View File

@ -56,7 +56,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
// 处理广播事件
if (content != null && (content.IsToMq || @event.IsToMq))
_broadcastingService.TrackChangePublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode);
_ = EnsureWriteAfterPublicTopicAsync(@event, content.SubSystemCode, content.TrackStatusCode);
if (canNextOrderIds.Count == 0) return;
// 根据订单关联关系异步向下刷（异步）
@ -277,7 +277,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
else
{
Debug.Assert(@event.CompareFields != null, "@event.CompareFields != null");
var extraDict = GetExtraDictForTextJson(@event.Extra);
var extraDict = TextJsonHelper.DeserializeDictionary(@event.Extra);
if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
{
throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}");
@ -285,7 +285,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
existTracks = existTracks.Where(x =>
{
var oldExtraDict = GetExtraDictForTextJson(x.Extra);
var oldExtraDict = TextJsonHelper.DeserializeDictionary(x.Extra);
// tips：注意兼容性，历史轨迹的字段可能为空
if (oldExtraDict == null || !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue))
{
@ -432,13 +432,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return track;
}
private Dictionary<string, object?>? GetExtraDictForTextJson(string? extraStr)
{
return string.IsNullOrWhiteSpace(extraStr)
? null
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(extraStr);
}
private OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order)
{
return new OrderOperationRecordIndex
@ -452,4 +445,16 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
CreateTime = DateTime.Now
};
}
private async Task EnsureWriteAfterPublicTopicAsync(TrackUpsertEvent @event, string subSystemCode, string statusCode)
{
//理想情况在写入之后1s会刷盘 https://www.elastic.co/guide/en/elasticsearch/reference/master/indices-refresh.html
//增加一个版本号来确定是否真的变更了
//没有刷盘成功事件回调
//方案1：!!【轮询等待静默刷盘】查出版本号确实发生了变更在发送广播
//方案2：考虑强制刷盘（对es压力大）后发送
//方案3：采用一个较大延时，不查询直接发
await Task.Delay(1000);
_broadcastingService.TrackChangePublicTopic(@event, subSystemCode, statusCode);
}
}

6
src/SJZY.Track.Application/Helpers/TextJsonHelper.cs
View File

@ -18,11 +18,11 @@ public static class TextJsonHelper
return string.IsNullOrWhiteSpace(json) ? default : System.Text.Json.JsonSerializer.Deserialize<T>(json);
}
public static Dictionary<string, object?>? DeserializeDictionary(string? dictString)
public static Dictionary<string, object?>? DeserializeDictionary(string? dictJsonString)
{
return string.IsNullOrWhiteSpace(dictString)
return string.IsNullOrWhiteSpace(dictJsonString)
? null
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictString);
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictJsonString);
}
/// <summary>

6
src/SJZY.Track.Application/Services/Broadcasting/BroadcastingService.cs
View File

@ -26,7 +26,7 @@ public class BroadcastingService : IBroadcastingService
/// <param name="subSystemCode">子系统代码</param>
/// <param name="trackStatusCode">轨迹状态代码</param>
/// <returns></returns>
public void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode)
public void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode)
{
var data = new TrackChangeEto
{
@ -65,7 +65,7 @@ public class BroadcastingService : IBroadcastingService
Encoding.UTF8.GetBytes(data.ToJson()));
}
public void TrackChangePublicTopicAsync(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode)
public void TrackChangePublicTopic(TrackDeleteEvent eto, OrderIdentifier[] orderIdentifiers, string trackCode, string trackStatusCode, string subSystemCode)
{
var data = new TrackChangeEto
{
@ -81,7 +81,7 @@ public class BroadcastingService : IBroadcastingService
Encoding.UTF8.GetBytes(data.ToJson()));
}
public void TrackChangePublicTopicAsync(TrackChangeEto eto, string subSystemCode)
public void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode)
{
_rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
Encoding.UTF8.GetBytes(eto.ToJson()));