YunTongJackYunTask/Reportapi/MyCode.Project.Services/Implementation/WorkProcessService.cs

using Microsoft.Practices.Unity;
using MyCode.Project.Domain.Config;
using MyCode.Project.Domain.Model;
using MyCode.Project.Domain.Repositories;
using MyCode.Project.Infrastructure.Common;
using MyCode.Project.Infrastructure.Enumeration;
using MyCode.Project.Infrastructure.Exceptions;
using MyCode.Project.Infrastructure.Extensions;
using MyCode.Project.Infrastructure.UnityExtensions;
using MyCode.Project.Repositories.Common;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace MyCode.Project.Services.Implementation
{
    public class WorkProcessService : ServiceBase, IWorkProcessService
    {
        private ISysWorkProcessRepository _SysWorkProcessRepository;

        private static object locker = new object(); //创建锁
        public WorkProcessService(ISysWorkProcessRepository SysWorkProcessRepository)
        {
            _SysWorkProcessRepository = SysWorkProcessRepository;
        }


        #region Add(添加调度)
        /// <summary>
        /// 添加调度
        /// </summary>
        /// <typeparam name="T">执行类</typeparam>
        /// <param name="merchantId">商家ID</param>
        /// <param name="methodName">方法名</param>
        /// <param name="remark">备注</param>
        /// <param name="entity">参数信息</param>
        /// <param name="funcType">执行类型</param>
        public void Add<T>(Guid merchantId, string methodName, string remark = "", object entity = null, int priority = 5, FuncType funcType = FuncType.Method) where T : class
        {
            Add(merchantId, typeof(T), methodName, remark, entity, funcType, priority);
        }

        /// <summary>
        /// 添加调度
        /// </summary>
        /// <param name="merchantId">商家ID</param>
        /// <param name="type">执行类</param>
        /// <param name="methodName">方法名</param>
        /// <param name="remark">备注</param>
        /// <param name="entity">参数信息</param>
        /// <param name="funcType">执行类型</param>
        public void Add(Guid merchantId, Type type, string methodName, string remark = "", object entity = null, FuncType funcType = FuncType.Method, int priority = 5)
        {
            string typePath = string.Format("{0}, {1}", type.FullName, type.Assembly.GetName().Name);
            string paramInfo = entity == null
                ? ""
                : entity is string || entity is Guid || entity is int || entity is long || entity is decimal ||
                  entity is float || entity is double
                    ? entity.SafeString()
                    : JsonHelper.ToJson(entity);
            Add(merchantId, typePath, methodName, remark, paramInfo, funcType, priority);
        }

        /// <summary>
        /// 添加调度任务
        /// </summary>
        /// <param name="merchantId">商家ID</param>
        /// <param name="typePath">类型路径，如：Lxm.IServices.IWorkProcessService, Lxm.Services</param>
        /// <param name="methodName">方法名</param>
        /// <param name="remark">备注</param>
        /// <param name="paramInfo">参数信息</param>
        /// <param name="funcType">执行类型</param>
        public void Add(Guid merchantId, string typePath, string methodName, string remark = "", string paramInfo = "",
            FuncType funcType = FuncType.Method, int priority = 5)
        {
            if (this._SysWorkProcessRepository.Exists(funcType, merchantId, typePath, methodName, paramInfo) && methodName != "RunWechatVSKingDee")
            {
                return;
            }
            SysWorkProcess entity = new SysWorkProcess();
            entity.ID = Guid.NewGuid();
            entity.MerchantID = merchantId;
            entity.FuncType = funcType.Value();
            entity.FuncClass = typePath;
            entity.FuncMethod = methodName;
            entity.ParamInfo = paramInfo;
            entity.FuncStatus = FuncStatus.Waiting.Value();
            entity.Remark = remark;
            entity.RetryCount = 0;
            entity.Creater = "系统调度";
            entity.CreateTime = DateTime.Now;
            entity.Editor = "系统调度";
            entity.EditTime = DateTime.Now;
            entity.Priority = priority;
            this._SysWorkProcessRepository.Add(entity);
        }

        #endregion



        #region SelectInitWorkProcess（返回前几十条数据）
        public List<SysWorkProcess> SelectInitWorkProcess(int top)
        {

            return _SysWorkProcessRepository.Queryable()
                .Where(p => p.FuncStatus == (int)WorkProcessStatus.Waiting && (p.Priority == 1 || p.Priority == null))
                .Take(top)
                .OrderBy(p => p.Priority)
                .OrderBy(p => p.CreateTime).ToList();
        }
        #endregion

        #region Add(批量添加)
        public void Add(List<SysWorkProcess> workProcess)
        {
            _SysWorkProcessRepository.Add(workProcess);
        }
        #endregion

        #region RestratStopProcess(重新启用所有暂停了的调度，这里不需要事务，因为修改失败也不影响)
        public void RestratStopProcess()
        {
            var list = _SysWorkProcessRepository.SelectList(p => p.FuncStatus == (int)WorkProcessStatus.Pause);
            list.ForEach(x => {
                x.FuncStatus = (int)WorkProcessStatus.Running;
                x.EditTime = DateTime.Now;
            });
            _SysWorkProcessRepository.Update(list);
        }
        #endregion

        #region RestartStopProcess（重新启用某个暂停了的调度）
        [TransactionCallHandler]
        public void RestartStopProcess(Guid workprocessId)
        {
            var workprocess = _SysWorkProcessRepository.SelectFirst(p => p.ID == workprocessId);

            if (workprocess.FuncStatus != (int)WorkProcessStatus.Pause) { throw new BaseException("当前进程状态不是停止"); }

            workprocess.FuncStatus = (int)WorkProcessStatus.Running;
            workprocess.EditTime = DateTime.Now;
            _SysWorkProcessRepository.Update(workprocess);

        }
        #endregion

        #region 清空历史数据
        public void ClearHistoryTask()
        {
            var historyDate = DateTime.Now.AddDays(-7);

            _SysWorkProcessRepository.Delete(p => p.FuncStatus == (int)WorkProcessStatus.Complete && p.EditTime < historyDate);
        }
        #endregion

        #region ExecuteSingle(执行单个)
        public void ExecuteSingle(SysWorkProcess process)
        {

            var type = UnityHelper.GetUnityContainer().Resolve(Type.GetType(process.FuncClass));

            MethodInfo method = type.GetType().GetMethod(process.FuncMethod);
            object result = new object();

            if (!string.IsNullOrEmpty(process.ParamInfo))
            {
                result= method.Invoke(type, new object[] { process.ParamInfo });
            }
            else
            {
                result= method.Invoke(type, new object[] { });
            }
            if (result == null)
                result = "";
            process.FuncStatus = (int)WorkProcessStatus.Complete;
            process.ExecuteTime = DateTime.Now;
            process.ExceptionInfo = string.Empty;
            process.EditTime = DateTime.Now;             
            process.Result=JsonHelper.ToJson(result);
            _SysWorkProcessRepository.Update(process);

        }
        #endregion

        #region Execute（调度执行）

        public void Execute()
        {
            lock (locker)
            {
                var client = UnityHelper.GetService<MyCodeSqlSugarClient>();

                var list = SelectInitWorkProcess(5);

                //先将这10个任务改成运行中
                if (list != null && list.Count > 0)
                {
                    var updateList = list.Select(p => new SysWorkProcess { ID = p.ID, FuncStatus = (int)WorkProcessStatus.Running });

                    _SysWorkProcessRepository.Update(updateList.ToList(), it => new { it.FuncStatus });
                }

                foreach (var process in list)
                {
                    //这里开启事务，同时才开启
                    client.Ado.BeginTran();

                    #region 执行一个任务
                    try
                    {
                        ExecuteSingle(process);

                        client.Ado.CommitTran();
                    }
                    catch (Exception ex)
                    {
                        client.Ado.RollbackTran();

                        while (ex.InnerException != null)
                        {
                            ex = ex.InnerException;
                        }

                        process.FuncStatus = (int)WorkProcessStatus.ExceptionStop;
                        process.EditTime = DateTime.Now;
                        process.ExecuteTime = DateTime.Now;
                        if (ex is BaseException)
                        {
                            process.ExceptionInfo = ex.Message;
                        }
                        else
                        {
                            process.ExceptionInfo = ex.ToString();
                        }

                        _SysWorkProcessRepository.Update(process);
                    }
                    #endregion
                }
            }
        }
        #endregion

        #region SelectOtherInitWorkProcess（返回优先级大于2的前几十条数据）
        /// <summary>
        /// 获取优先级3，4,5的调度
        /// </summary>
        /// <param name="top"></param>
        /// <returns></returns>
        public List<SysWorkProcess> SelectOtherInitWorkProcess(int top)
        {

            return _SysWorkProcessRepository.Queryable()
                .Where(p => p.FuncStatus == (int)WorkProcessStatus.Waiting && p.Priority > 2 && p.Priority <= 5)
                .Take(top)
                .OrderBy(p => p.Priority)
                .OrderBy(p => p.CreateTime).ToList();
        }
        #endregion

        #region ExecuteOther（调度执行优先级比较低的任务）
        /// <summary>
        /// 调度执行优先级比较低的任务
        /// </summary>
        public void ExecuteOther()
        {
            var client = UnityHelper.GetService<MyCodeSqlSugarClient>();

            var list = SelectOtherInitWorkProcess(20);

            //先将这10个任务改成运行中
            if (list != null && list.Count > 0)
            {
                var updateList = list.Select(p => new SysWorkProcess { ID = p.ID, FuncStatus = (int)WorkProcessStatus.Running });

                _SysWorkProcessRepository.Update(updateList.ToList(), it => new { it.FuncStatus });
            }

            foreach (var process in list)
            {
                //这里开启事务，同时才开启
                client.Ado.BeginTran();

                #region 执行一个任务
                try
                {
                    ExecuteSingle(process);

                    client.Ado.CommitTran();
                }
                catch (Exception ex)
                {
                    client.Ado.RollbackTran();

                    while (ex.InnerException != null)
                    {
                        ex = ex.InnerException;
                    }

                    process.FuncStatus = (int)WorkProcessStatus.ExceptionStop;
                    process.EditTime = DateTime.Now;
                    process.ExecuteTime = DateTime.Now;
                    if (ex is BaseException)
                    {
                        process.ExceptionInfo = ex.Message;
                    }
                    else
                    {
                        process.ExceptionInfo = ex.ToString();
                    }

                    _SysWorkProcessRepository.Update(process);
                }
                #endregion
            }
        }
        #endregion

        #region SelectPriority2InitWorkProcess（返回优先级等于2的前几十条数据）
        /// <summary>
        /// 返回优先级等于2的前几十条数据
        /// </summary>
        /// <param name="top"></param>
        /// <returns></returns>
        public List<SysWorkProcess> SelectPriority2InitWorkProcess(int top)
        {

            return _SysWorkProcessRepository.Queryable()
                .Where(p => p.FuncStatus == (int)WorkProcessStatus.Waiting && p.Priority == 2)
                .Take(top)
                .OrderBy(p => p.Priority)
                .OrderBy(p => p.CreateTime).ToList();
        }
        #endregion

        #region ExecutePriority2Work（调度执行优先级等于2的任务）
        /// <summary>
        /// 调度执行优先级等于2的任务
        /// </summary>
        public void ExecutePriority2Work()
        {
            var client = UnityHelper.GetService<MyCodeSqlSugarClient>();

            var list = SelectPriority2InitWorkProcess(10);

            //先将这10个任务改成运行中
            if (list != null && list.Count > 0)
            {
                var updateList = list.Select(p => new SysWorkProcess { ID = p.ID, FuncStatus = (int)WorkProcessStatus.Running });

                _SysWorkProcessRepository.Update(updateList.ToList(), it => new { it.FuncStatus });
            }

            foreach (var process in list)
            {
                //这里开启事务，同时才开启
                client.Ado.BeginTran();

                #region 执行一个任务
                try
                {
                    ExecuteSingle(process);

                    client.Ado.CommitTran();
                }
                catch (Exception ex)
                {
                    client.Ado.RollbackTran();

                    while (ex.InnerException != null)
                    {
                        ex = ex.InnerException;
                    }

                    process.FuncStatus = (int)WorkProcessStatus.ExceptionStop;
                    process.EditTime = DateTime.Now;
                    process.ExecuteTime = DateTime.Now;
                    if (ex is BaseException)
                    {
                        process.ExceptionInfo = ex.Message;
                    }
                    else
                    {
                        process.ExceptionInfo = ex.ToString();
                    }

                    _SysWorkProcessRepository.Update(process);
                }
                #endregion
                //Thread.Sleep(300);


            }
        }
        #endregion

        #region SelectSmsInitWorkProcess（获取优先级6的短信调度）
        /// <summary>
        /// 获取优先级6的短信调度
        /// </summary>
        /// <param name="top"></param>
        /// <returns></returns>
        public List<SysWorkProcess> SelectPriority6WorkProcess(int top)
        {
            DateTime days = DateTime.Now.Date;
            return _SysWorkProcessRepository.Queryable()
                .Where(p => p.FuncStatus == (int)WorkProcessStatus.Waiting && p.Priority == 6)
                .Take(top)
                .OrderBy(p => p.Priority)
                .OrderBy(p => p.CreateTime).ToList();
        }
        #endregion

        #region ExecutePriority6（调度执行优先级=6的任务）
        /// <summary>
        /// 调度执行优先级=6的任务
        /// </summary>
        public void ExecutePriority6()
        {
            var client = UnityHelper.GetService<MyCodeSqlSugarClient>();

            var list = SelectPriority6WorkProcess(20);

            //先将这10个任务改成运行中
            if (list != null && list.Count > 0)
            {
                var updateList = list.Select(p => new SysWorkProcess { ID = p.ID, FuncStatus = (int)WorkProcessStatus.Running });

                _SysWorkProcessRepository.Update(updateList.ToList(), it => new { it.FuncStatus });
            }

            foreach (var process in list)
            {
                //这里开启事务，同时才开启
                client.Ado.BeginTran();

                #region 执行一个任务
                try
                {
                    ExecuteSingle(process);

                    client.Ado.CommitTran();
                }
                catch (Exception ex)
                {
                    client.Ado.RollbackTran();

                    while (ex.InnerException != null)
                    {
                        ex = ex.InnerException;
                    }

                    process.FuncStatus = (int)WorkProcessStatus.ExceptionStop;
                    process.EditTime = DateTime.Now;
                    process.ExecuteTime = DateTime.Now;
                    if (ex is BaseException)
                    {
                        process.ExceptionInfo = ex.Message;
                    }
                    else
                    {
                        process.ExceptionInfo = ex.ToString();
                    }

                    _SysWorkProcessRepository.Update(process);
                }
                #endregion
            }
        }
        #endregion

        #region RetryTask(重试失败的任务)
        /// <summary>
        /// 重试失败的任务
        /// </summary>
        public void RetryTask()
        {
            DateTime today = DateTime.Now.Date.AddDays(-1);
            var list = _SysWorkProcessRepository.Queryable().Where(t => t.FuncStatus == 4 && t.RetryCount <= 10
             && t.CreateTime >= today).OrderBy(t => t.EditTime).Take(20).ToList();
            list.ForEach(t =>
            {
                t.RetryCount = t.RetryCount + 1;
                t.FuncStatus = 0;
            });
            _SysWorkProcessRepository.Update(list);
        }
        #endregion
    }
}