tnb.server/WarehouseMgr/Tnb.WarehouseMgr/TimedTaskBackgroundService.cs

using System.Collections.Concurrent;
using Aop.Api.Domain;
using JNPF;
using JNPF.Common.Extension;
using JNPF.Common.Security;
using JNPF.EventBus;
using JNPF.EventHandler;
using JNPF.Systems.Entitys.System;
using Microsoft.Extensions.Hosting;
using NPOI.OpenXmlFormats.Dml.Diagram;
using Tnb.Common.Extension;
using Tnb.Common.Utils;
using Tnb.WarehouseMgr.Entities.Configs;
using Tnb.WarehouseMgr.Entities.Consts;
using Tnb.WarehouseMgr.Entities.Enums;
using Tnb.WarehouseMgr.Entities.Exceptions;

namespace Tnb.WarehouseMgr
{
    /// <summary>
    /// 定时任务
    /// added by ly on 20230802
    /// </summary>
    public class TimedTaskBackgroundService : BackgroundService
    {
        public bool IsStarted { get; set; }
        private IEventPublisher _eventPublisher = default!;
        private readonly ElevatorControlConfiguration _elevatorControlConfiguration = App.Configuration.Build<ElevatorControlConfiguration>();
        private readonly IServiceProvider _serviceProvider;
        private readonly IHostApplicationLifetime _lifeTime;
        private List<string> s_heartbeatDevNames = new();
        //private static Dictionary<string, Func<CancellationToken?, Task>> _timedFuncMap = new(StringComparer.OrdinalIgnoreCase);
        public TimedTaskBackgroundService()
        {
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            IsStarted = true;
            s_heartbeatDevNames = _elevatorControlConfiguration.HeartbeatDevNames;
            Task timedTask = Task.Run(() =>
             {
                 _eventPublisher = App.GetRequiredService<IEventPublisher>();
                 //电梯Agv心跳检测
                 _ = TimedTask(async token =>
                 {
                     foreach (var devName in s_heartbeatDevNames)
                     {
                         Dictionary<string, string> parameter = new()
                         {
                             ["DevName"] = devName,
                             ["TagName"] = ElevatorConsts.AGVKeepalive,
                             ["Value"] = "123",
                             ["token"] = _elevatorControlConfiguration.token
                         };
                         string result = await HttpClientHelper.GetAsync(_elevatorControlConfiguration.WriteTagUrl, pars: parameter);

                         //var result = await RedisHelper.HSetAsync(devName, ElevatorConsts.AGVKeepalive, "123");
                         await Console.Out.WriteLineAsync($"{devName.Match(@"\d+")}#梯, 心跳检测结果:{result}");
                     }
                 }, stoppingToken, 30);

             });
            return timedTask;
        }



        private Task TimedTask(Func<CancellationToken, Task> action, CancellationToken ct, int interval, TimeSpanUnit timeType = TimeSpanUnit.Seconds)
        {
            //    CancellationToken token = ct;
            //    return Task.Factory.StartNew(async () =>
            //    {
            //        while (!token.IsCancellationRequested)
            //        {
            //            await action(ct).Catch(async ex =>
            //            {

            //                if (ex is TimedTaskException timedTaskEx and not null)
            //                {
            //                    await _eventPublisher.PublishAsync(new LogEventSource("Log:CreateExLog", timedTaskEx.options!, new SysLogEntity
            //                    {
            //                        Id = SnowflakeIdHelper.NextId(),
            //                        Category = 4,
            //                        UserId = timedTaskEx.UserId,
            //                        UserName = timedTaskEx.UserName,
            //                        IPAddress = NetHelper.Ip,
            //                        RequestURL = timedTaskEx.RequestURL,
            //                        RequestMethod = timedTaskEx.RequestMethod,
            //                        Json = timedTaskEx + "\n" + timedTaskEx.InnerException?.StackTrace + "\n" + timedTaskEx?.TargetSite?.GetParameters().ToString(),
            //                        //PlatForm = string.Format("{0}-{1}", userAgent.OS.ToString(), userAgent.RawValue),
            //                        CreatorTime = DateTime.Now
            //                    }));
            //                }
            //            });
            //            await TaskDelay(timeType, interval);
            //        }
            //    }, ct, TaskCreationOptions.None, new CustomerTaskScheduler());

            return Task.Run(async () =>
            {
                while (!ct.IsCancellationRequested)
                {

                    await action(ct).Catch(async ex =>
                    {
                        if (ex is TimedTaskException timedTaskEx and not null)
                        {
                            await _eventPublisher.PublishAsync(new LogEventSource("Log:CreateExLog", timedTaskEx.options!, new SysLogEntity
                            {
                                Id = SnowflakeIdHelper.NextId(),
                                Category = 4,
                                UserId = timedTaskEx.UserId,
                                UserName = timedTaskEx.UserName,
                                IPAddress = NetHelper.Ip,
                                RequestURL = timedTaskEx.RequestURL,
                                RequestMethod = timedTaskEx.RequestMethod,
                                Json = timedTaskEx + "\n" + timedTaskEx.InnerException?.StackTrace + "\n" + timedTaskEx?.TargetSite?.GetParameters().ToString(),
                                //PlatForm = string.Format("{0}-{1}", userAgent.OS.ToString(), userAgent.RawValue),
                                CreatorTime = DateTime.Now
                            }));
                        }
                    });
                    await TaskDelay(timeType, interval);
                }
            }, ct);
        }

        public override Task StopAsync(CancellationToken cancellationToken)
        {

            IsStarted = false;
            return Task.CompletedTask;
        }

        private Task TaskDelay(TimeSpanUnit timeType, int interval)
        {
            Task delayTask = timeType switch
            {
                TimeSpanUnit.Milliseconds => Task.Delay(TimeSpan.FromMilliseconds(interval)),
                TimeSpanUnit.Seconds => Task.Delay(TimeSpan.FromSeconds(interval)),
                TimeSpanUnit.Minutes => Task.Delay(TimeSpan.FromMinutes(interval)),
                TimeSpanUnit.Hours => Task.Delay(TimeSpan.FromHours(interval)),
                TimeSpanUnit.Days => Task.Delay(TimeSpan.FromDays(interval)),
                _ => throw new NotImplementedException()
            };
            return delayTask;
        }

        public Task CloseAgvHeartbeat(IEnumerable<string> devNames)
        {
            s_heartbeatDevNames.RemoveAll(x => devNames.Contains(x));
            return Task.CompletedTask;
        }

        public Task OpenAgvHeartbeat(IEnumerable<string> devNames)
        {
            s_heartbeatDevNames.AddRange(devNames);
            return Task.CompletedTask;
        }

    }
    /// <summary>
    /// 自定义任务调度器，保证长任务在单独的线程中运行
    /// </summary>
    internal class CustomerTaskScheduler : TaskScheduler
    {
        // 这边的 BlockingCollection 只是举个例子，如果是普通的队列，配合锁也是可以的。
        private readonly BlockingCollection<Task> _tasks = new();

        public CustomerTaskScheduler()
        {
            Thread thread = new(() =>
            {
                foreach (Task task in _tasks.GetConsumingEnumerable())
                {
                    _ = TryExecuteTask(task);
                }
            })
            {
                IsBackground = true
            };
            thread.Start();
        }

        protected override IEnumerable<Task> GetScheduledTasks()
        {
            return _tasks;
        }

        protected override void QueueTask(Task task)
        {
            _tasks.Add(task);
        }

        protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        {
            return false;
        }
    }

}