Merge remote-tracking branch 'origin/S03-轨迹中台优化项' into test

This commit is contained in:
dingenwen 2024-05-14 17:23:44 +08:00
commit 2e67b13756
3 changed files with 44 additions and 33 deletions

View File

@ -21,6 +21,8 @@
<PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.1.2"/> <PackageReference Include="DotNetCore.CAP.Dashboard" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.1.2"/> <PackageReference Include="DotNetCore.CAP.RabbitMQ" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.MySql" Version="8.1.2"/> <PackageReference Include="DotNetCore.CAP.MySql" Version="8.1.2"/>
<PackageReference Include="DotNetCore.CAP.InMemoryStorage" Version="8.1.2"/>
<PackageReference Include="Savorboard.CAP.InMemoryMessageQueue" Version="7.0.0"/>
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>

View File

@ -78,46 +78,50 @@ public class TrackPullService : ApplicationService
/// <returns></returns> /// <returns></returns>
[AllowAnonymous] [AllowAnonymous]
[HttpPost("Track/RefreshForTask")] [HttpPost("Track/RefreshForTask")]
public async Task<bool> RefreshForTaskAsync() public bool RefreshForTaskAsync()
{ {
var configs = await _trackPullCacheService.GetAllPullConfigsAsync(); Task.Run(async () =>
foreach (var config in configs)
{ {
var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId); var configs = await _trackPullCacheService.GetAllPullConfigsAsync();
if (await RedisHelper.Client.ExistsAsync(redisKey))
{
continue;
}
var page = 1; foreach (var config in configs)
var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{ {
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode; var redisKey = _trackPullCacheService.TrackPullKey(config.SubSystemId, config.ServiceProviderId);
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName; if (await RedisHelper.Client.ExistsAsync(redisKey))
foreach (var group in orders.GroupBy(x =>
new { x.SubSystemId, x.OrderType }))
{ {
var data = group.Select(x => new continue;
{
x.OrderNo
}).ToList();
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
serviceProviderName!,
group.Key.SubSystemId, group.Key.OrderType));
// 批量获取轨迹
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach);
} }
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page); var page = 1;
} var orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, page);
while (orders.Any())
{
var serviceProviderCode = orders.FirstOrDefault()?.ServiceProviderCode;
var serviceProviderName = orders.FirstOrDefault()?.ServiceProviderName;
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count); foreach (var group in orders.GroupBy(x =>
} new { x.SubSystemId, x.OrderType }))
{
var data = group.Select(x => new
{
x.OrderNo
}).ToList();
var attach = Base64Helper.SerializeObject(new TrackRefreshAttach(config.ServiceProviderId,
serviceProviderName!,
group.Key.SubSystemId, group.Key.OrderType));
// 批量获取轨迹
await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync(
InterfaceRelayConst.GetTrackInterfaceCode, serviceProviderCode, data,
GlobalConst.TrackRefreshSubscribe, attach);
}
orders = await _provideQueries.FindOrderWithServiceProviderAsync(config.ServiceProviderId, ++page);
}
await RedisHelper.Client.SetExAsync(redisKey, config.PullInterval * 60, orders.Count);
}
});
return true; return true;
} }

View File

@ -1,5 +1,6 @@
using DotNetCore.CAP; using DotNetCore.CAP;
using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Configuration;
using Savorboard.CAP.InMemoryMessageQueue;
using SJZY.Expand.ABP.Core.Localization; using SJZY.Expand.ABP.Core.Localization;
using SJZY.Expand.ABP.Core.Logger; using SJZY.Expand.ABP.Core.Logger;
using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.Expand.ABP.Core.PublicMethords;
@ -76,6 +77,10 @@ namespace SJZY.Track.Application
context.Services.AddCapTableNameService(); context.Services.AddCapTableNameService();
context.Services.AddCap(options => context.Services.AddCap(options =>
{ {
#if DEBUG
options.UseInMemoryStorage();
options.UseInMemoryMessageQueue();
#else
options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!); options.UseMySql(configuration["ConnectionStrings:DefaultConnection"]!);
options.UseRabbitMQ(configure => options.UseRabbitMQ(configure =>
{ {
@ -86,7 +91,7 @@ namespace SJZY.Track.Application
configure.BasicQosOptions = configure.BasicQosOptions =
new RabbitMQOptions.BasicQos(configuration.GetValue<ushort>("CAP:PrefetchCount")); new RabbitMQOptions.BasicQos(configuration.GetValue<ushort>("CAP:PrefetchCount"));
}); });
// options.TopicNamePrefix = "v1"; #endif
options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval"); options.FailedRetryInterval = configuration.GetValue<int>("CAP:FailedRetryInterval");
options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount"); options.FailedRetryCount = configuration.GetValue<int>("CAP:FailedRetryCount");
options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter"); options.SucceedMessageExpiredAfter = configuration.GetValue<int>("CAP:SucceedMessageExpiredAfter");