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

Merge branch 'S06' into test

Browse Source
This commit is contained in:
hardstylewyl 2025-01-16 14:27:43 +08:00
commit 3cec26133d
7 changed files with 105 additions and 169 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="eto"></param>
/// <param name="subSystemCode"></param> /// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param> /// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode); void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode);
/// <summary> /// <summary>
/// 轨迹变更广播（删除） /// 轨迹变更广播（删除）
@ -20,7 +20,7 @@ public interface IBroadcastingService : ITransientDependency
/// <param name="eto"></param> /// <param name="eto"></param>
/// <param name="subSystemCode"></param> /// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></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> /// <summary>
/// 轨迹变更广播（删除） /// 轨迹变更广播（删除）
@ -28,7 +28,7 @@ public interface IBroadcastingService : ITransientDependency
/// <param name="eto"></param> /// <param name="eto"></param>
/// <param name="subSystemCode"></param> /// <param name="subSystemCode"></param>
/// <param name="trackStatusCode"></param> /// <param name="trackStatusCode"></param>
void TrackChangePublicTopicAsync(TrackChangeEto eto, string subSystemCode); void TrackChangePublicTopic(TrackChangeEto eto, string subSystemCode);
/// <summary> /// <summary>
/// 轨迹推送完成广播（推送完成回调） /// 轨迹推送完成广播（推送完成回调）

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

@ -47,7 +47,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
//当轨迹删除，根据轨迹标记/状态上IsToMq选择进行广播 //当轨迹删除，根据轨迹标记/状态上IsToMq选择进行广播
if (trackList.Count != 0) if (trackList.Count != 0)
await PublicTrackTopicAsync(@event, trackList); _ = PublicTrackTopicAsync(@event, trackList);
// 根据订单关联关系异步向下刷（异步） // 根据订单关联关系异步向下刷（异步）
await DownByOrderLink(@event); await DownByOrderLink(@event);
@ -121,89 +121,6 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
return true; return true;
} }
private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List<TrackDetailsIndex> trackList)
{
var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray();
var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray();
//从轨迹内容中查询出标记是否需要mq广播
var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes);
var toMqDict = contentList
.ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq);
//选出需要广播的轨迹,并按code和订单号分组
var trackGroup = trackList
.Where(x => toMqDict[x.SubSystemId + x.TrackCode])
.GroupBy(m => new
{
m.TrackCode,
m.OrderNo
})
.ToList();
if (trackGroup.Count == 0) return;
foreach (var dataItem in trackGroup)
{
var item = dataItem.First();
var content = contentList.First(m => m.TrackContentCode == item.TrackCode);
// _broadcastingService.TrackChangePublicTopicAsync(track,
// dataItem.Select(m => new OrderIdentifier
// {
// OrderId = m.OrderId,
// OrderNo = m.OrderNo,
// OrderType = m.OrderType
// }).ToArray(),
// item.TrackCode,
// item.TrackStatusCode,
// content.SubSystemCode
// );
//将每个订单的每个轨迹的删除变更发布
foreach (var track in dataItem)
{
_broadcastingService.TrackChangePublicTopicAsync(new TrackChangeEto
{
EstimatedDeliveryDate = track.EstimatedDeliveryDate,
TrackPlacePostcode = track.TrackPlacePostcode,
TrackPlaceNation = track.TrackPlaceNation,
TrackPlaceNationId = track.TrackPlaceNationId,
TrackPlaceProvince = track.TrackPlaceProvince,
TrackPlaceProvinceId = track.TrackPlaceProvinceId,
TrackPlaceCity = track.TrackPlaceCity,
TrackPlaceCityId = track.TrackPlaceCityId,
Files = track.Files?.Select(x => new FileItemDto()
{
FileName = x.FileName,
OssKey = x.OssKey
}).ToList(),
Remarks = track.Remarks,
Source = track.Source,
IsVisibility = track.IsVisibility,
TrackTimeZone = track.TrackTimeZone,
TrackTime = track.TrackTime,
TrackPlace = track.TrackPlace,
TrackStatusCode = track.TrackStatusCode,
EnStatus = track.TrackEnContent,
Content = track.TrackContent,
TrackCode = track.TrackCode,
OperationMan = @event.OperationMan,
OrderIdentifiers = new[]
{
new OrderIdentifier()
{
OrderId = track.OrderId,
OrderNo = track.OrderNo,
OrderType = track.OrderType
}
},
SubSystemId = @event.SubSystemId,
Extra = track.Extra
}, content.SubSystemCode);
}
}
}
private async Task<List<TrackDetailsIndex>> DeleteTrackAsync(TrackDeleteEvent @event) private async Task<List<TrackDetailsIndex>> DeleteTrackAsync(TrackDeleteEvent @event)
{ {
List<int> orderIds; List<int> orderIds;
@ -264,7 +181,7 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
throw new BusinessException(message: $"当前轨迹不存在扩展信息，无法利用扩展字段：{@event.CompareFields}"); throw new BusinessException(message: $"当前轨迹不存在扩展信息，无法利用扩展字段：{@event.CompareFields}");
} }
var extraDict = GetExtraDictForTextJson(x.Extra); var extraDict = TextJsonHelper.DeserializeDictionary(x.Extra);
if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue)) if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
{ {
throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}"); throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}");
@ -361,68 +278,88 @@ public class TrackDeleteEventHandler : ITransientDependency, ICapSubscribe
return deleteTracks; return deleteTracks;
} }
private Dictionary<string, object?>? GetExtraDictForTextJson(string? extraStr) private async Task PublicTrackTopicAsync(TrackDeleteEvent @event, List<TrackDetailsIndex> trackList)
{ {
return string.IsNullOrWhiteSpace(extraStr) var systemIds = trackList.Select(x => x.SubSystemId).Distinct().ToArray();
? null var trackCodes = trackList.Select(x => x.TrackCode).Distinct().ToArray();
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(extraStr);
}
//json Element和object的比较 //从轨迹内容中查询出标记是否需要mq广播
private bool EqJsonElementObject2(object? a, object? b) var contentList = await _provideQueries.GetTrackContentsAsync(systemIds, trackCodes);
{ var toMqDict = contentList
if (a == null && b == null) .ToDictionary(x => x.SubSystemId + x.TrackContentCode, y => y.IsToMq);
{
return true;
}
if (a is JsonElement aJsonElement) //选出需要广播的轨迹,并按code和订单号分组
{ var trackGroup = trackList
switch (aJsonElement.ValueKind) .Where(x => toMqDict[x.SubSystemId + x.TrackCode])
.GroupBy(m => new
{ {
case JsonValueKind.Undefined: m.TrackCode,
return false; m.OrderNo
case JsonValueKind.Object: })
return false; .ToList();
case JsonValueKind.Array: if (trackGroup.Count == 0) return;
return false;
case JsonValueKind.String: //增加延时确保写盘成功
return string.Equals(aJsonElement.GetString(), b as string); await Task.Delay(1000);
case JsonValueKind.Number: foreach (var dataItem in trackGroup)
{
var item = dataItem.First();
var content = contentList.First(m => m.TrackContentCode == item.TrackCode);
// _broadcastingService.TrackChangePublicTopic(track,
// dataItem.Select(m => new OrderIdentifier
// {
// OrderId = m.OrderId,
// OrderNo = m.OrderNo,
// OrderType = m.OrderType
// }).ToArray(),
// item.TrackCode,
// item.TrackStatusCode,
// content.SubSystemCode
// );
//将每个订单的每个轨迹的删除变更发布
foreach (var track in dataItem)
{
_broadcastingService.TrackChangePublicTopic(new TrackChangeEto
{
EstimatedDeliveryDate = track.EstimatedDeliveryDate,
TrackPlacePostcode = track.TrackPlacePostcode,
TrackPlaceNation = track.TrackPlaceNation,
TrackPlaceNationId = track.TrackPlaceNationId,
TrackPlaceProvince = track.TrackPlaceProvince,
TrackPlaceProvinceId = track.TrackPlaceProvinceId,
TrackPlaceCity = track.TrackPlaceCity,
TrackPlaceCityId = track.TrackPlaceCityId,
Files = track.Files?.Select(x => new FileItemDto()
{ {
if (b == null) return false; FileName = x.FileName,
OssKey = x.OssKey
if (b is int bInt) }).ToList(),
{ Remarks = track.Remarks,
var aInt = aJsonElement.GetInt32(); Source = track.Source,
return aInt == bInt; IsVisibility = track.IsVisibility,
} TrackTimeZone = track.TrackTimeZone,
TrackTime = track.TrackTime,
if (b is long bLong) TrackPlace = track.TrackPlace,
{ TrackStatusCode = track.TrackStatusCode,
var aLong = aJsonElement.GetInt64(); EnStatus = track.TrackEnContent,
return aLong == bLong; Content = track.TrackContent,
} TrackCode = track.TrackCode,
OperationMan = @event.OperationMan,
return false; OrderIdentifiers = new[]
}
case JsonValueKind.True:
case JsonValueKind.False:
{ {
if (b is bool bl) new OrderIdentifier()
{ {
return aJsonElement.GetBoolean() == bl; OrderId = track.OrderId,
OrderNo = track.OrderNo,
OrderType = track.OrderType
} }
},
return false; SubSystemId = @event.SubSystemId,
} Extra = track.Extra
case JsonValueKind.Null: }, content.SubSystemCode);
{
return false;
}
} }
} }
return false;
} }
} }

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

@ -408,7 +408,7 @@ public class TrackPushEventHandler : ITransientDependency, ICapSubscribe
if (config.IsAllPushWithLatest) if (config.IsAllPushWithLatest)
{ {
//同code同比较字段保证唯一 //同code同比较字段保证唯一
var old = body!.Data var old = body.Data
.Where(x => .Where(x =>
{ {
//反序列化出的JsonElement和字典字符串类型做比较 //反序列化出的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)) 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; if (canNextOrderIds.Count == 0) return;
// 根据订单关联关系异步向下刷（异步） // 根据订单关联关系异步向下刷（异步）
@ -277,7 +277,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
else else
{ {
Debug.Assert(@event.CompareFields != null, "@event.CompareFields != null"); 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)) if (extraDict == null || !extraDict.TryGetValue(@event.CompareFields, out var compareValue))
{ {
throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}"); throw new BusinessException(message: $"请检查拓展字段，为什么找不到比较字段：{@event.CompareFields}");
@ -285,7 +285,7 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
existTracks = existTracks.Where(x => existTracks = existTracks.Where(x =>
{ {
var oldExtraDict = GetExtraDictForTextJson(x.Extra); var oldExtraDict = TextJsonHelper.DeserializeDictionary(x.Extra);
// tips：注意兼容性，历史轨迹的字段可能为空 // tips：注意兼容性，历史轨迹的字段可能为空
if (oldExtraDict == null || !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue)) if (oldExtraDict == null || !oldExtraDict.TryGetValue(@event.CompareFields, out var oldValue))
{ {
@ -432,13 +432,6 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
return track; 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) private OrderOperationRecordIndex BuildOrderOperationRecord(TrackUpsertEvent @event, TrackDetailsIndex track, OrderIdentifier order)
{ {
return new OrderOperationRecordIndex return new OrderOperationRecordIndex
@ -452,4 +445,16 @@ public class TrackUpsertEventHandler : ITransientDependency, ICapSubscribe
CreateTime = DateTime.Now 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); 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 ? null
: System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictString); : System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object?>>(dictJsonString);
} }
/// <summary> /// <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="subSystemCode">子系统代码</param>
/// <param name="trackStatusCode">轨迹状态代码</param> /// <param name="trackStatusCode">轨迹状态代码</param>
/// <returns></returns> /// <returns></returns>
public void TrackChangePublicTopicAsync(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode) public void TrackChangePublicTopic(TrackUpsertEvent eto, string subSystemCode, string trackStatusCode)
{ {
var data = new TrackChangeEto var data = new TrackChangeEto
{ {
@ -65,7 +65,7 @@ public class BroadcastingService : IBroadcastingService
Encoding.UTF8.GetBytes(data.ToJson())); 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 var data = new TrackChangeEto
{ {
@ -81,7 +81,7 @@ public class BroadcastingService : IBroadcastingService
Encoding.UTF8.GetBytes(data.ToJson())); 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}", _rabbitClientBus.Publish(GlobalConst.TrackChangeExchange, $"TrackChange.{subSystemCode}",
Encoding.UTF8.GetBytes(eto.ToJson())); Encoding.UTF8.GetBytes(eto.ToJson()));

16
src/SJZY.Track.Application/Services/ProvideServices/TrackService.cs
View File

@ -255,11 +255,8 @@ public partial class TrackService : ApplicationService, ITrackService
} }
var extraDict = System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object>>(y.Extra); var extraDict = System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object>>(y.Extra);
if (extraDict!.TryGetValue(input.CompareFields!, out var compareValue)) if (extraDict!.TryGetValue(input.CompareFields!, out var compareValueJsonElement))
{ return TextJsonHelper.EqJsonElementAndObject(compareValueJsonElement, input.CompareValue);
//todo:需要修改JsonElement和object的逻辑
return compareValue.Equals(input.CompareValue);
}
return false; return false;
}) })
@ -378,12 +375,9 @@ public partial class TrackService : ApplicationService, ITrackService
throw new BusinessException(message: "轨迹extra为空无法使用比较字段比较，请检查轨迹"); throw new BusinessException(message: "轨迹extra为空无法使用比较字段比较，请检查轨迹");
} }
var extraDict = System.Text.Json.JsonSerializer.Deserialize<Dictionary<string, object>>(y.Extra); var extraDict = TextJsonHelper.DeserializeDictionary(y.Extra);
if (extraDict!.TryGetValue(input.CompareFields!, out var compareValue)) if (extraDict!.TryGetValue(input.CompareFields!, out var compareValueJsonElement))
{ return TextJsonHelper.EqJsonElementAndObject(compareValueJsonElement, input.CompareValue);
//todo:需要修改JsonElement和object的逻辑
return compareValue.Equals(input.CompareValue);
}
return false; return false;
}) })