Browse Source

Cron,增加 Break(Type) 方法,允许 Cron 打断自己的循环。

master
王厅 1 week ago
parent
commit
a0b786b6b2
  1. 248
      Apewer/CronAttribute.cs

248
Apewer/CronAttribute.cs

@ -1,5 +1,6 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.ComponentModel.Design;
using System.Reflection; using System.Reflection;
using System.Text; using System.Text;
using System.Threading; using System.Threading;
@ -8,10 +9,12 @@ namespace Apewer
{ {
/// <summary>Cron 特性,默认间隔为 60 秒。</summary> /// <summary>Cron 特性,默认间隔为 60 秒。</summary>
[AttributeUsage(AttributeTargets.Class, AllowMultiple = true, Inherited = false)] [AttributeUsage(AttributeTargets.Class, AllowMultiple = false, Inherited = false)]
public sealed class CronAttribute : Attribute, IToJson public sealed class CronAttribute : Attribute, IToJson
{ {
#region attribute
private int _mode = 0; private int _mode = 0;
private int _seconds = 0; private int _seconds = 0;
private CronCycle _cycle = CronCycle.Once; private CronCycle _cycle = CronCycle.Once;
@ -51,7 +54,23 @@ namespace Apewer
return json; return json;
} }
#region Payload /// <summary></summary>
public override string ToString()
{
if (_type != null)
{
if (_mode == 1) return $"Type = {_type.FullName}, Seconds = {_seconds}";
if (_mode == 2) return $"Type = {_type.FullName}, Cycle = {_cycle}, Seconds = {_seconds}";
}
return base.ToString();
}
#endregion
#region payload
/// <summary>执行 Cron 的类。</summary>
public Type Type { get => _type; }
// 传入。 // 传入。
Type _type = null; Type _type = null;
@ -61,13 +80,25 @@ namespace Apewer
// 实时。 // 实时。
Thread _thread = null; Thread _thread = null;
bool _alive = false; bool _alive = false;
bool _break = false;
/// <summary>运行当前 Cron。</summary>
void Run() void Run()
{ {
_alive = true; if (_alive) return;
_thread = new Thread(Payload); if (_type == null) return;
_thread.IsBackground = false;
_thread.Start(); _break = false;
UsePool(pool =>
{
if (pool.ContainsKey(_type)) return;
pool.Add(_type, this);
_alive = true;
_thread = new Thread(Payload);
_thread.IsBackground = false;
_thread.Start();
});
} }
// 当前进程睡眠。 // 当前进程睡眠。
@ -83,11 +114,11 @@ namespace Apewer
var next = DateTime.MinValue; var next = DateTime.MinValue;
while (true) while (true)
{ {
if (_break) break; if (_break || _break_all) break;
if (_now < next) if (_now < next)
{ {
Sleep(); Sleep();
if (_break) break; if (_break || _break_all) break;
continue; continue;
} }
@ -102,9 +133,9 @@ namespace Apewer
var next = DateTime.Now.AddSeconds(_seconds); var next = DateTime.Now.AddSeconds(_seconds);
while (_now < next) while (_now < next)
{ {
if (_break) break; if (_break || _break_all) break;
Sleep(); Sleep();
if (_break) break; if (_break || _break_all) break;
} }
Invoke(); Invoke();
} }
@ -134,11 +165,11 @@ namespace Apewer
// 启动循环。 // 启动循环。
while (true) while (true)
{ {
if (_break) break; if (_break || _break_all) break;
if (_now < next) if (_now < next)
{ {
Sleep(); Sleep();
if (_break) break; if (_break || _break_all) break;
continue; continue;
} }
@ -164,7 +195,13 @@ namespace Apewer
} }
} }
_alive = false; UsePool(pool =>
{
if (pool.ContainsKey(_type)) pool.Remove(_type);
_break = false;
_alive = false;
});
} }
void Invoke() void Invoke()
@ -188,13 +225,52 @@ namespace Apewer
#endregion #endregion
#region CronInvoker #region pool
static Dictionary<Type, CronAttribute> _pool = new Dictionary<Type, CronAttribute>();
static void UsePool(Action<Dictionary<Type, CronAttribute>> callback)
{
if (callback == null) throw new ArgumentNullException(nameof(callback));
lock (_pool)
{
callback.Invoke(_pool);
}
}
static object _start = new object(); static T UsePool<T>(Func<Dictionary<Type, CronAttribute>, T> callback)
static CronAttribute[] _crons = null; {
static bool _break = false; if (callback == null) throw new ArgumentNullException(nameof(callback));
lock (_pool)
{
return callback.Invoke(_pool);
}
}
#endregion
#region invoker
static bool _break_all = false;
static DateTime _now; static DateTime _now;
static int _crons_alive = 0;
/// <summary>获取状态,指示打断 Cron 循环。</summary>
public static bool Breaking { get => _break_all; }
/// <summary>正在运行的 Cron 数量。</summary>
public static int AliveCount { get => CountAlive(UsePool(pool => pool.Values)); }
/// <summary>获取正在运行的 Cron 数量。</summary>
/// <returns></returns>
static int CountAlive(IEnumerable<CronAttribute> crons)
{
var alive = 0;
foreach (var cron in crons)
{
if (cron._alive) alive += 1;
}
return alive;
}
static CronAttribute[] Init(IEnumerable<Assembly> assemblies, Logger logger, bool logEvent) static CronAttribute[] Init(IEnumerable<Assembly> assemblies, Logger logger, bool logEvent)
{ {
@ -226,6 +302,14 @@ namespace Apewer
return array; return array;
} }
/// <summary>启动指定的 Cron。</summary>
/// <exception cref="ArgumentNullException"></exception>
public static void Start(CronAttribute cron)
{
if (cron == null) throw new ArgumentNullException(nameof(cron));
cron.Run();
}
/// <summary>开始 Cron 调用(阻塞当前线程)。</summary> /// <summary>开始 Cron 调用(阻塞当前线程)。</summary>
/// <param name="assemblies">包含 Cron 的程序集,不指定此参数时将在 AppDomain 中搜索。</param> /// <param name="assemblies">包含 Cron 的程序集,不指定此参数时将在 AppDomain 中搜索。</param>
/// <param name="logger">日志记录程序,不指定此参数时将使用 Logger.Default。</param> /// <param name="logger">日志记录程序,不指定此参数时将使用 Logger.Default。</param>
@ -239,71 +323,122 @@ namespace Apewer
public static void Start(IEnumerable<Assembly> assemblies = null, Logger logger = null, bool logEvent = true) public static void Start(IEnumerable<Assembly> assemblies = null, Logger logger = null, bool logEvent = true)
{ {
if (logger == null) logger = Logger.Default; if (logger == null) logger = Logger.Default;
lock (_start)
// 设置控制台事件。
Console.CancelKeyPress += (s, e) =>
{ {
// 初始化。 _break_all = true;
_now = DateTime.Now; e.Cancel = true;
_crons = Init(assemblies, logger, logEvent); };
if (_crons.Length < 1)
{ // 初始化。
logger.Error(nameof(CronAttribute), "没有找到带有 Cron 特性的类型。"); _now = DateTime.Now;
return; var crons = Init(assemblies, logger, logEvent);
} if (crons.Length > 0)
{
logger.Text(nameof(CronAttribute), "Crons started。");
// 启动线程。 // 启动 Cron。
_crons_alive = _crons.Length; foreach (var cron in crons) Start(cron);
Console.CancelKeyPress += (s, e) =>
{
_break = true;
e.Cancel = true;
};
logger.Text(nameof(CronAttribute), $"启动 {_crons.Length} 个 Cron 线程。");
foreach (var cron in _crons) cron.Run();
// 监视退出状态 // 等待 Cron 退出。
while (true) while (true)
{ {
Thread.Sleep(100); Thread.Sleep(100);
_now = DateTime.Now; _now = DateTime.Now;
var alive = 0; if (AliveCount < 1) break;
for (var i = 0; i < _crons.Length; i++) if (_crons[i]._alive) alive += 1;
if (alive < 1) break;
_crons_alive = alive;
} }
logger.Text(nameof(CronAttribute), "所有 Cron 已结束。");
} }
logger.Text(nameof(CronAttribute), "Crons ended。");
} }
/// <summary>打断正在执行的 Cron。</summary> /// <summary>打断指定的 Cron。</summary>
public static void Abort() /// <exception cref="ArgumentNullException" />
public static void Abort(Type type)
{ {
var crons = _crons; if (type == null) throw new ArgumentNullException(nameof(type));
if (crons != null)
UsePool(pool =>
{ {
foreach (var cron in crons) if (pool.TryGetValue(type, out var cron))
{ {
if (cron == null) continue;
try { cron._thread.Abort(); } catch { } try { cron._thread.Abort(); } catch { }
} }
}
pool.Remove(type);
});
}
/// <summary>打断指定的 Cron。</summary>
/// <exception cref="ArgumentNullException" />
public static void Abort(object instance)
{
if (instance == null) throw new ArgumentNullException(nameof(instance));
if (instance is Type type) Abort(type);
else Abort(instance.GetType());
} }
/// <summary>打断 Cron 循环,不打断正在执行的 Cron。</summary> /// <summary>打断所有 Cron。</summary>
public static void Abort()
{
UsePool(pool =>
{
foreach (var item in pool)
{
try
{
item.Value._thread.Abort();
item.Value._alive = false;
item.Value._break = false;
}
catch { }
}
});
}
/// <summary>打断 Cron 循环,不打断正在执行的 Job。</summary>
/// <exception cref="ArgumentNullException" />
public static void Break(Type type)
{
if (type == null) throw new ArgumentNullException(nameof(type));
UsePool(pool =>
{
if (pool.TryGetValue(type, out var cron))
{
cron._break = true;
}
});
}
/// <summary>打断 Cron 循环,不打断正在执行的 Job。</summary>
/// <exception cref="ArgumentNullException" />
public static void Break(object instance)
{
if (instance == null) throw new ArgumentNullException(nameof(instance));
if (instance is Type type) Break(type);
else Break(instance.GetType());
}
/// <summary>打断 Cron 循环,不打断正在执行的 Job。</summary>
public static void Break() public static void Break()
{ {
_break = true; _break_all = true;
} }
/// <summary>打断 Cron 循环并等待 Cron 执行结束。等待指定的时间后打断正在执行的 Cron。</summary> /// <summary>打断 Cron 循环并等待 Job 执行结束。等待指定的时间后打断正在执行的 Job。</summary>
/// <param name="timeout">强制打断前的等待毫秒数,指定为负数时将无限等待。</param> /// <param name="timeout">强制打断前的等待毫秒数,指定为负数时将无限等待。</param>
public static void Break(int timeout) public static void Break(int timeout)
{ {
_break = true; _break_all = true;
const int interval = 100; const int interval = 100;
if (timeout < 0) if (timeout < 0)
{ {
while (_crons_alive > 0) Thread.Sleep(interval); while (AliveCount > 0) Thread.Sleep(interval);
return; return;
} }
else else
@ -311,7 +446,7 @@ namespace Apewer
if (timeout > 0) if (timeout > 0)
{ {
var remains = timeout; var remains = timeout;
while (remains > 0 && _crons_alive > 0) while (remains > 0 && AliveCount > 0)
{ {
Thread.Sleep(interval); Thread.Sleep(interval);
remains -= interval; remains -= interval;
@ -321,9 +456,6 @@ namespace Apewer
} }
} }
/// <summary>获取状态,指示打断 Cron 循环。</summary>
public static bool Breaking { get => _break; }
#endregion #endregion
} }

Loading…
Cancel
Save