wenkai.chen/tnb.server
Template
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

继续调整，手动开启定时任务相关逻辑代码，并处理相关bug

Browse Source
This commit is contained in:
alex
2023-09-20 18:20:15 +08:00
parent 4220ca4b8b
commit 8ce7bb540e
12 changed files with 53 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
WarehouseMgr/Tnb.WarehouseMgr.Interfaces/IWareHouseService.cs
View File

@@ -52,7 +52,7 @@ namespace Tnb.WarehouseMgr.Interfaces
/// 生成任务执行
/// </summary>
/// <returns></returns>
Task GenTaskExecute(CancellationTokenSource? cts = default);
Task GenTaskExecute(CancellationToken? ct = default);
/// <summary>
/// 任务完成
/// </summary>

2
WarehouseMgr/Tnb.WarehouseMgr.Interfaces/IWmsPDATransferSignService.cs
View File

@@ -11,6 +11,6 @@ namespace Tnb.WarehouseMgr.Interfaces
/// </summary>
public interface IWmsPDATransferSignService
{
Task IsMinStorage(CancellationTokenSource? cts = default);
Task IsMinStorage(CancellationToken? cts = default);
}
}

2
WarehouseMgr/Tnb.WarehouseMgr.Interfaces/IWmsSetSortingService.cs
View File

@@ -11,6 +11,6 @@ namespace Tnb.WarehouseMgr.Interfaces
/// </summary>
public interface IWmsSetSortingService
{
Task PackSortingByAdd(CancellationTokenSource? cts = default);
Task PackSortingByAdd(CancellationToken? cts = default);
}
}

4
WarehouseMgr/Tnb.WarehouseMgr.Interfaces/IWmskittingOutService.cs
View File

@@ -22,12 +22,12 @@ namespace Tnb.WarehouseMgr.Interfaces
/// 齐套出库（新增状态）
/// </summary>
/// <returns></returns>
Task KittingOutByAdd(CancellationTokenSource? cts = default);
Task KittingOutByAdd(CancellationToken? ct = default);
/// <summary>
/// 齐套出库，（待配送状态）
/// </summary>
/// <param name="cts"></param>
/// <returns></returns>
Task KittingOutByIsToBeShipped(CancellationTokenSource? cts = default);
Task KittingOutByIsToBeShipped(CancellationToken? ct = default);
}
}

4
WarehouseMgr/Tnb.WarehouseMgr/PDATransferSignService.cs
View File

@@ -149,7 +149,7 @@ namespace Tnb.WarehouseMgr
/// </summary>
/// <returns></returns>
[HttpPost]
public async Task IsMinStorage(CancellationTokenSource? cts = default)
public async Task IsMinStorage(CancellationToken? cts = default)
{
//if (UserManager.AsscessToken.IsNullOrWhiteSpace()) return;
//var curUser = await GetUserIdentity();
@@ -224,7 +224,7 @@ namespace Tnb.WarehouseMgr
userIdentity = await GetUserIdentity(_userManager.ToKen),
};
var timedTaskEx = ex.ToTimedTaskException(ei);
cts?.Cancel();
//cts?.Cancel();
throw timedTaskEx;
}
}

31
WarehouseMgr/Tnb.WarehouseMgr/TimedTaskBackgroundService.cs
View File

@@ -42,7 +42,7 @@ namespace Tnb.WarehouseMgr
public bool IsStarted { get; set; }
private IEventPublisher _eventPublisher = default!;
private readonly IServiceProvider _serviceProvider;
private static Dictionary<string, Func<CancellationTokenSource?, Task>> _timedFuncMap = new(StringComparer.OrdinalIgnoreCase);
private static Dictionary<string, Func<CancellationToken?, Task>> _timedFuncMap = new(StringComparer.OrdinalIgnoreCase);
static TimedTaskBackgroundService()
{
Task.Run(() =>
@@ -50,7 +50,7 @@ namespace Tnb.WarehouseMgr
_timedFuncMap = App.EffectiveTypes.AsParallel().Where(t => !t.Namespace.IsNullOrWhiteSpace() && t.Namespace.Equals("Tnb.WarehouseMgr", StringComparison.OrdinalIgnoreCase)).SelectMany(t => t.GetMethods())
.Where(m => m.GetCustomAttribute<TimedAttribute>() != null)
.ToDictionary(x => x.Name, x =>
(Func<CancellationTokenSource?, Task>)Delegate.CreateDelegate(typeof(Func<CancellationTokenSource?, Task>), App.GetService(x.DeclaringType), x));
(Func<CancellationToken?, Task>)Delegate.CreateDelegate(typeof(Func<CancellationToken?, Task>), App.GetService(x.DeclaringType), x));
});
}
public TimedTaskBackgroundService(IServiceProvider serviceProvider)
@@ -74,7 +74,7 @@ namespace Tnb.WarehouseMgr
{
if (_timedFuncMap.ContainsKey(message.TaskName))
{
await _timedFuncMap[message.TaskName].Invoke(cts);
await _timedFuncMap[message.TaskName].Invoke(stoppingToken);
}
}
}
@@ -83,24 +83,16 @@ namespace Tnb.WarehouseMgr
var timedTask = Task.Run(() =>
{
_eventPublisher = App.GetRequiredService<IEventPublisher>();
////生成任务执行
//CancellationTokenSource genTaskCTS = new();
CancellationTokenSource kittingOutAddCts = new();
CancellationTokenSource kittingOutShippedCts = new();
CancellationTokenSource setSortingCts = new();
CancellationTokenSource isMinStorageCts = new();
//var wareHouseService = App.GetRequiredService<IWareHouseService>();
//TimedTask(cts => wareHouseService.GenTaskExecute(cts), genTaskCTS);
//齐套出库
var kittingOutService = App.GetRequiredService<IWmskittingOutService>();
TimedTask(cts => kittingOutService.KittingOutByAdd(cts), kittingOutAddCts, 1);
TimedTask(cts => kittingOutService.KittingOutByIsToBeShipped(cts), kittingOutShippedCts, 1);
TimedTask(cts => kittingOutService.KittingOutByAdd(stoppingToken), stoppingToken, 1);
TimedTask(cts => kittingOutService.KittingOutByIsToBeShipped(stoppingToken), stoppingToken, 1);
//齐套分拣
var setSortingService = App.GetRequiredService<IWmsSetSortingService>();
TimedTask(cts => setSortingService.PackSortingByAdd(cts), setSortingCts, 1);
TimedTask(cts => setSortingService.PackSortingByAdd(cts), stoppingToken, 1);
//最低库存检查
var transferSignService = App.GetRequiredService<IWmsPDATransferSignService>();
TimedTask(cts => transferSignService.IsMinStorage(cts), isMinStorageCts, 30, TimeSpanUnit.Minutes);
TimedTask(cts => transferSignService.IsMinStorage(cts), stoppingToken, 30, TimeSpanUnit.Minutes);
}, stoppingToken);
@@ -109,14 +101,14 @@ namespace Tnb.WarehouseMgr
private Task TimedTask(Func<CancellationTokenSource, Task> action, CancellationTokenSource cts, int interval, TimeSpanUnit timeType = TimeSpanUnit.Seconds)
private Task TimedTask(Func<CancellationToken, Task> action, CancellationToken ct, int interval, TimeSpanUnit timeType = TimeSpanUnit.Seconds)
{
var token = cts.Token;
var token = ct;
return Task.Factory.StartNew(async () =>
{
while (!token.IsCancellationRequested)
{
await action(cts).Catch(async ex =>
await action(ct).Catch(async ex =>
{
if (ex is TimedTaskException timedTaskEx and not null)
{
@@ -137,7 +129,7 @@ namespace Tnb.WarehouseMgr
});
await TaskDelay(timeType, interval);
}
}, cts.Token, TaskCreationOptions.None, new CustomerTaskScheduler());
}, ct, TaskCreationOptions.None, new CustomerTaskScheduler());
#region ThreadPool 线程运行会导致线程饥饿
//return Task.Run(async () =>
@@ -172,6 +164,7 @@ namespace Tnb.WarehouseMgr
public override Task StopAsync(CancellationToken cancellationToken)
{
IsStarted = false;
return Task.CompletedTask;
//return base.StopAsync(cancellationToken);

9
WarehouseMgr/Tnb.WarehouseMgr/WareHouseService.cs
View File

@@ -211,9 +211,10 @@ namespace Tnb.WarehouseMgr
whereExprable.And(carryStatusFilterExp);
var whereExpr = whereExprable.ToExpression();
var policy = await _db.Queryable<WmsInstockPolicies>().Where(it => it.status == 1).FirstAsync();
var cyDb = _db.CopyNew();
var policy = await cyDb.Queryable<WmsInstockPolicies>().Where(it => it.status == 1).FirstAsync();
if (policy == null) throw new AppFriendlyException("没有可用策略", 500);
var items = await _db.Queryable<WmsCarryH>().LeftJoin<WmsCarryCode>((a, b) => a.id == b.carry_id)
var items = await cyDb.Queryable<WmsCarryH>().LeftJoin<WmsCarryCode>((a, b) => a.id == b.carry_id)
.LeftJoin<BasLocation>((a, b, c) => a.location_id == c.id)
.Where(whereExpr)
.OrderBy(policy.policy)
@@ -226,7 +227,7 @@ namespace Tnb.WarehouseMgr
/// </summary>
/// <returns></returns>
[HttpPost, Timed(Name = nameof(IWareHouseService.GenTaskExecute))]
public async Task GenTaskExecute(CancellationTokenSource? cts = default)
public async Task GenTaskExecute(CancellationToken? ct = default)
{
Stopwatch sw = Stopwatch.StartNew();
CancellationTokenSource agvCts = new();
@@ -378,13 +379,11 @@ namespace Tnb.WarehouseMgr
{
Log.Error("生成预任务执行时出现错误", ex);
await db.Ado.RollbackTranAsync();
cts?.Cancel();
throw;
}
finally
{
agvCts.Dispose();
cts?.Dispose();
}
}
/// <summary>

4
WarehouseMgr/Tnb.WarehouseMgr/WmsSetSortingService.cs
View File

@@ -95,7 +95,7 @@ namespace Tnb.WarehouseMgr
/// </summary>
/// <returns></returns>
[HttpPost, Timed(Name = nameof(PackSortingByAdd))]
public async Task PackSortingByAdd(CancellationTokenSource? cts = default)
public async Task PackSortingByAdd(CancellationToken? ct = default)
{
//if (UserManager.AsscessToken.IsNullOrWhiteSpace()) return;
//var curUser = await GetUserIdentity();
@@ -270,7 +270,7 @@ namespace Tnb.WarehouseMgr
userIdentity = await GetUserIdentity(_userManager.ToKen),
};
var timedTaskEx = ex.ToTimedTaskException(ei);
cts?.Cancel();
//cts?.Cancel();
throw timedTaskEx;
}
finally

12
WarehouseMgr/Tnb.WarehouseMgr/WmskittingOutService.cs
View File

@@ -78,10 +78,14 @@ namespace Tnb.WarehouseMgr
/// </summary>
/// <returns></returns>
[HttpPost, Timed(Name = nameof(KittingOutByAdd))]
public async Task KittingOutByAdd(CancellationTokenSource? cts = default)
public async Task KittingOutByAdd(CancellationToken? ct = default)
{
//if (UserManager.AsscessToken.IsNullOrWhiteSpace()) return;
//var curUser = await GetUserIdentity();
if (ct?.IsCancellationRequested ?? false)
{
ct?.ThrowIfCancellationRequested();
}
var curDb = _db.CopyNew();
try
{
@@ -174,7 +178,7 @@ namespace Tnb.WarehouseMgr
userIdentity = await GetUserIdentity(_userManager.ToKen),
};
var timedTaskEx = ex.ToTimedTaskException(ei);
cts?.Cancel();
//cts?.Cancel();
throw timedTaskEx;
}
}
@@ -183,7 +187,7 @@ namespace Tnb.WarehouseMgr
/// </summary>
/// <returns></returns>
[HttpPost, Timed(Name = nameof(KittingOutByIsToBeShipped))]
public async Task KittingOutByIsToBeShipped(CancellationTokenSource? cts = default)
public async Task KittingOutByIsToBeShipped(CancellationToken? ct = default)
{
//if (UserManager.AsscessToken.IsNullOrWhiteSpace()) return;
//var curUser = await GetUserIdentity();
@@ -281,7 +285,7 @@ namespace Tnb.WarehouseMgr
userIdentity = await GetUserIdentity(_userManager.ToKen),
};
var timedTaskEx = ex.ToTimedTaskException(ei);
cts?.Cancel();
//cts?.Cancel();
throw timedTaskEx;
}
finally

6
common/Tnb.Common.Core/Manager/User/UserManager.cs
View File

@@ -76,7 +76,8 @@ public partial class UserManager : IUserManager, IScoped
/// </summary>
public UserEntity User
{
get =>_repository.GetSingle(u => u.Id == UserId);
//modify by ly on 20230920
get => _repository.CopyNew().Queryable<UserEntity>().Single(u => u.Id == UserId); //_repository.GetSingle(u => u.Id == UserId);
}
/// <summary>
@@ -230,7 +231,8 @@ public partial class UserManager : IUserManager, IScoped
var ipAddress = NetHelper.Ip;
//var ipAddressName = await NetHelper.GetLocation(ipAddress);
var sysConfigInfo = await _repository.AsSugarClient().Queryable<SysConfigEntity>().FirstAsync(s => s.Category.Equals("SysConfig") && s.Key.ToLower().Equals("tokentimeout"));
data = await _repository.AsQueryable().Where(it => it.Id == UserId)
var db = _repository.CopyNew();
data = await db.Queryable<UserEntity>().Where(it => it.Id == UserId)
.Select(a => new UserInfoModel
{
userId = a.Id,

17
system/Tnb.OAuth/OAuthService.cs
View File

@@ -145,7 +145,7 @@ public class OAuthService : IDynamicApiController, ITransient
private readonly IMHandler _imHandler;
private readonly BackgroundService _backgroundService; //added by ly on 20230916
private static CancellationTokenSource stopTimedTaskSvcCTS = new();
private static CancellationTokenSource stopTimedTaskSvcCTS;
private static Dictionary<string, Func<BackgroundService, bool>> _fetchPropValue = new();
@@ -296,18 +296,26 @@ public class OAuthService : IDynamicApiController, ITransient
//modify by ly on 20230918 登录成功后启动定时服务
if (!userId.IsNullOrWhiteSpace())
{
if (!_fetchPropValue.TryGetValue("IsStarted", out var action))
{
var isStartedProp = _backgroundService.GetType().GetProperty("IsStarted");
var paramExp = Expression.Parameter(typeof(BackgroundService), "_backgroundSvc");
var propExp = Expression.Property(Expression.ConvertChecked(paramExp, isStartedProp.DeclaringType), isStartedProp.GetGetMethod());
var propExp = Expression.Property(Expression.ConvertChecked(paramExp, isStartedProp.DeclaringType), isStartedProp);
var body = Expression.Lambda<Func<BackgroundService, Boolean>>(propExp, paramExp);
action = body.Compile();
_fetchPropValue["IsStarted"] = action;
}
if (!action?.Invoke(_backgroundService) ?? false)
{
await _backgroundService.StartAsync(stopTimedTaskSvcCTS.Token);
stopTimedTaskSvcCTS = new();
try
{
_backgroundService.StartAsync(stopTimedTaskSvcCTS.Token);
}
catch (TaskCanceledException)
{
}
}
}
@@ -411,8 +419,11 @@ public class OAuthService : IDynamicApiController, ITransient
public async Task Logout([FromQuery] string ticket)
{
//modify by ly on 20230918
stopTimedTaskSvcCTS ??= new();
await _backgroundService.StopAsync(stopTimedTaskSvcCTS.Token);
stopTimedTaskSvcCTS.Cancel();
//stopTimedTaskSvcCTS.Dispose();
UserManager.AsscessToken = string.Empty;
var tenantId = _userManager.TenantId ?? "default";

3
taskschedule/Tnb.TaskScheduler/Listener/SpareTimeListener.cs
View File

@@ -21,7 +21,7 @@ public class SpareTimeListener : ISpareTimeListener, ISingleton
/// 构造函数.
/// </summary>
public SpareTimeListener(
ISqlSugarClient context,
ISqlSugarClient context,
IEventPublisher eventPublisher)
{
_eventPublisher = eventPublisher;
@@ -88,6 +88,7 @@ public class SpareTimeListener : ISpareTimeListener, ISingleton
//modified by PhilPan
var nextRunTime = SpareTime.GetCronNextOccurrence(taskEntity.ExecuteCycleJson)?.DateTime;
await _eventPublisher.PublishAsync(new TaskEventSource("Task:UpdateTask", connectionConfig, new TimeTaskEntity()
{
Id = taskEntity.Id,