NewLife
/
AntJob
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

v3.2.2023.0610 进一步完善埋点

This commit is contained in:
大石头 2023-06-10 08:56:49 +08:00
parent 3e110d9d1b
commit fddb030f2f
26 changed files with 1754 additions and 1826 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
AntJob.Agent/Program.cs
View File

@ -1,49 +1,40 @@
using System;
using System.Threading;
using AntJob;
using AntJob.Extensions;
using AntJob.Providers;
using NewLife.Log;
using NewLife.Model;
using Stardust;
namespace AntJob.Agent;
// 启用控制台日志，拦截所有异常
XTrace.UseConsole();
class Program
var services = ObjectContainer.Current;
services.AddStardust();
var set = AntSetting.Current;
// 实例化调度器
var scheduler = new Scheduler
{
static void Main(String[] args)
ServiceProvider = services.BuildServiceProvider(),
// 使用分布式调度引擎替换默认的本地文件调度
Provider = new NetworkJobProvider
{
// 启用控制台日志，拦截所有异常
XTrace.UseConsole();
var services = ObjectContainer.Current;
var star = new StarFactory();
var set = AntSetting.Current;
// 实例化调度器
var scheduler = new Scheduler
{
ServiceProvider = services.BuildServiceProvider(),
// 使用分布式调度引擎替换默认的本地文件调度
Provider = new NetworkJobProvider
{
Debug = set.Debug,
Server = set.Server,
AppID = set.AppID,
Secret = set.Secret,
}
};
// 添加作业处理器
//sc.Handlers.Add(new CSharpHandler());
scheduler.AddHandler<SqlHandler>();
scheduler.AddHandler<SqlMessage>();
// 启动调度引擎，调度器内部多线程处理
scheduler.Start();
// 友好退出
//ObjectContainer.Current.BuildHost().Run();
Thread.Sleep(-1);
Debug = set.Debug,
Server = set.Server,
AppID = set.AppID,
Secret = set.Secret,
}
}
};
// 添加作业处理器
//sc.Handlers.Add(new CSharpHandler());
scheduler.AddHandler<SqlHandler>();
scheduler.AddHandler<SqlMessage>();
// 启动调度引擎，调度器内部多线程处理
scheduler.Start();
// 友好退出
services.BuildHost().Run();

4
AntJob.Extensions/AntJob.Extensions.csproj
View File

@ -1,12 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard2.1;netstandard2.0;net461</TargetFrameworks>
<TargetFrameworks>netstandard2.1;netstandard2.0;net461;net45</TargetFrameworks>
<AssemblyTitle>蚂蚁数据调度SDK</AssemblyTitle>
<Description>分布式任务调度系统，纯NET打造的重量级大数据实时计算平台，万亿级调度经验积累。</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2023 NewLife</Copyright>
<VersionPrefix>3.2</VersionPrefix>
<VersionPrefix>3.3</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
<FileVersion>$(Version)</FileVersion>

353
AntJob.Extensions/DataHandler.cs
View File

@ -4,197 +4,192 @@ using NewLife;
using XCode;
using XCode.Configuration;
namespace AntJob
namespace AntJob.Extensions;
/// <summary>数据处理作业（泛型）</summary>
/// <remarks>
/// 定时调度只要达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// 任务切片条件：StartTime + Step + Offset &lt;= Now
/// </remarks>
/// <typeparam name="TEntity"></typeparam>
public abstract class DataHandler<TEntity> : DataHandler where TEntity : Entity<TEntity>, new()
{
/// <summary>数据处理作业（泛型）</summary>
/// <remarks>
/// 定时调度只要达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// 任务切片条件：StartTime + Step + Offset &lt;= Now
/// </remarks>
/// <typeparam name="TEntity"></typeparam>
public abstract class DataHandler<TEntity> : DataHandler where TEntity : Entity<TEntity>, new()
/// <summary>实例化数据处理作业</summary>
public DataHandler() => Factory = Entity<TEntity>.Meta.Factory;
#region 数据处理
/// <summary>处理一批数据</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
/// <summary>实例化数据处理作业</summary>
public DataHandler() => Factory = Entity<TEntity>.Meta.Factory;
var count = 0;
foreach (var item in ctx.Data as IEnumerable)
if (ProcessItem(ctx, item as TEntity)) count++;
#region 数据处理
/// <summary>处理一批数据</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var count = 0;
foreach (var item in ctx.Data as IEnumerable)
{
if (ProcessItem(ctx, item as TEntity)) count++;
}
return count;
}
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="entity"></param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, TEntity entity) => true;
#endregion
return count;
}
/// <summary>数据处理作业</summary>
/// <remarks>
/// 定时调度只要达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// 任务切片条件：StartTime + Step + Offset &lt;= Now
/// </remarks>
public abstract class DataHandler : Handler
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="entity"></param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, TEntity entity) => true;
#endregion
}
/// <summary>数据处理作业</summary>
/// <remarks>
/// 定时调度只要达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// 任务切片条件：StartTime + Step + Offset &lt;= Now
/// </remarks>
public abstract class DataHandler : Handler
{
#region 属性
/// <summary>实体工厂</summary>
public IEntityFactory Factory { get; set; }
/// <summary>附加条件</summary>
public String Where { get; set; }
/// <summary>时间字段 或 雪花Id</summary>
public FieldItem Field { get; set; }
/// <summary>排序</summary>
public String OrderBy { get; set; }
/// <summary>选择列</summary>
public String Selects { get; set; }
#endregion
#region 构造
/// <summary>实例化数据库处理器</summary>
public DataHandler() => Mode = JobModes.Data;
#endregion
#region 方法
/// <summary>开始</summary>
public override Boolean Start()
{
#region 属性
/// <summary>实体工厂</summary>
public IEntityFactory Factory { get; set; }
if (Active) return false;
/// <summary>附加条件</summary>
public String Where { get; set; }
if (Factory == null) throw new ArgumentNullException(nameof(Factory));
/// <summary>时间字段 或 雪花Id</summary>
public FieldItem Field { get; set; }
/// <summary>排序</summary>
public String OrderBy { get; set; }
/// <summary>选择列</summary>
public String Selects { get; set; }
#endregion
#region 构造
/// <summary>实例化数据库处理器</summary>
public DataHandler() => Mode = JobModes.Data;
#endregion
#region 方法
/// <summary>开始</summary>
public override Boolean Start()
// 自动识别雪花Id字段
if (Field == null)
{
if (Active) return false;
if (Factory == null) throw new ArgumentNullException(nameof(Factory));
// 自动识别雪花Id字段
if (Field == null)
{
var pks = Factory.Table.PrimaryKeys;
if (pks != null && pks.Length == 1 && pks[0].Type == typeof(Int64)) Field = pks[0];
}
if (Field == null) Field = Factory.MasterTime;
if (Field == null) throw new ArgumentNullException(nameof(Field));
var job = Job;
if (job.Step == 0) job.Step = 30;
// 获取最小时间
if (job.Start.Year < 2000) throw new InvalidOperationException("数据任务必须设置开始时间");
return base.Start();
var pks = Factory.Table.PrimaryKeys;
if (pks != null && pks.Length == 1 && pks[0].Type == typeof(Int64)) Field = pks[0];
}
#endregion
Field ??= Factory.MasterTime;
if (Field == null) throw new ArgumentNullException(nameof(Field));
#region 数据处理
/// <summary>处理任务。内部分批处理</summary>
/// <param name="ctx"></param>
protected override void OnProcess(JobContext ctx)
{
var prov = Provider;
var row = 0;
while (true)
{
ctx.Data = null;
//ctx.Entity = null;
ctx.Error = null;
var job = Job;
if (job.Step == 0) job.Step = 30;
// 分批抽取
var data = Fetch(ctx, ref row);
// 获取最小时间
if (job.Start.Year < 2000) throw new InvalidOperationException("数据任务必须设置开始时间");
var list = data as IList;
if (list != null) ctx.Total += list.Count;
ctx.Data = data;
if (data == null || list != null && list.Count == 0) break;
// 报告进度
ctx.Status = JobStatus.处理中;
prov?.Report(ctx);
// 批量处理
ctx.Success += Execute(ctx);
// 报告进度
ctx.Status = JobStatus.抽取中;
// 不满一批，结束
if (list != null && list.Count < ctx.Task.BatchSize) break;
}
}
/// <summary>分批抽取数据，一个任务内多次调用</summary>
/// <param name="ctx">上下文</param>
/// <param name="row">开始行数</param>
/// <returns></returns>
protected virtual Object Fetch(JobContext ctx, ref Int32 row)
{
var task = ctx.Task;
if (task == null) throw new ArgumentNullException(nameof(task), "没有设置数据抽取配置");
// 验证时间段
var start = task.Start;
var end = task.End;
// 区间无效
if (start >= end) return null;
// 分批获取数据，如果没有取到，则结束
var fi = Field;
var exp = new WhereExpression();
if (fi.Type == typeof(DateTime))
{
if (start > DateTime.MinValue && start < DateTime.MaxValue) exp &= fi >= start;
if (end > DateTime.MinValue && end < DateTime.MaxValue) exp &= fi < end;
}
else if (fi.Type == typeof(Int64))
{
var snow = Factory.Snow;
if (start > DateTime.MinValue && start < DateTime.MaxValue) exp &= fi >= snow.GetId(start);
if (end > DateTime.MinValue && end < DateTime.MaxValue) exp &= fi < snow.GetId(end);
}
else
throw new NotSupportedException($"不支持抽取[{fi.Type.FullName}]类型的字段数据！");
if (!Where.IsNullOrEmpty()) exp &= Where;
var list = Factory.FindAll(exp, OrderBy, Selects, row, task.BatchSize);
// 取到数据，需要滑动窗口
if (list.Count > 0) row += list.Count;
return list;
}
/// <summary>处理一批数据</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var count = 0;
foreach (var item in ctx.Data as IEnumerable)
{
if (ProcessItem(ctx, item as IEntity)) count++;
}
return count;
}
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="entity"></param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, IEntity entity) => true;
#endregion
return base.Start();
}
#endregion
#region 数据处理
/// <summary>处理任务。内部分批处理</summary>
/// <param name="ctx"></param>
protected override void OnProcess(JobContext ctx)
{
var prov = Provider;
var row = 0;
while (true)
{
ctx.Data = null;
//ctx.Entity = null;
ctx.Error = null;
// 分批抽取
var data = Fetch(ctx, ref row);
var list = data as IList;
if (list != null) ctx.Total += list.Count;
ctx.Data = data;
if (data == null || list != null && list.Count == 0) break;
// 报告进度
ctx.Status = JobStatus.处理中;
prov?.Report(ctx);
// 批量处理
ctx.Success += Execute(ctx);
// 报告进度
ctx.Status = JobStatus.抽取中;
// 不满一批，结束
if (list != null && list.Count < ctx.Task.BatchSize) break;
}
}
/// <summary>分批抽取数据，一个任务内多次调用</summary>
/// <param name="ctx">上下文</param>
/// <param name="row">开始行数</param>
/// <returns></returns>
protected virtual Object Fetch(JobContext ctx, ref Int32 row)
{
var task = ctx.Task;
if (task == null) throw new ArgumentNullException(nameof(task), "没有设置数据抽取配置");
// 验证时间段
var start = task.Start;
var end = task.End;
// 区间无效
if (start >= end) return null;
// 分批获取数据，如果没有取到，则结束
var fi = Field;
var exp = new WhereExpression();
if (fi.Type == typeof(DateTime))
{
if (start > DateTime.MinValue && start < DateTime.MaxValue) exp &= fi >= start;
if (end > DateTime.MinValue && end < DateTime.MaxValue) exp &= fi < end;
}
else if (fi.Type == typeof(Int64))
{
var snow = Factory.Snow;
if (start > DateTime.MinValue && start < DateTime.MaxValue) exp &= fi >= snow.GetId(start);
if (end > DateTime.MinValue && end < DateTime.MaxValue) exp &= fi < snow.GetId(end);
}
else
throw new NotSupportedException($"不支持抽取[{fi.Type.FullName}]类型的字段数据！");
if (!Where.IsNullOrEmpty()) exp &= Where;
var list = Factory.FindAll(exp, OrderBy, Selects, row, task.BatchSize);
// 取到数据，需要滑动窗口
if (list.Count > 0) row += list.Count;
return list;
}
/// <summary>处理一批数据</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var count = 0;
foreach (var item in ctx.Data as IEnumerable)
if (ProcessItem(ctx, item as IEntity)) count++;
return count;
}
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="entity"></param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, IEntity entity) => true;
#endregion
}

181
AntJob.Extensions/SqlHandler.cs
View File

@ -1,109 +1,98 @@
using System;
using AntJob.Data;
using AntJob.Extensions;
using AntJob.Data;
using NewLife.Data;
using XCode.DataAccessLayer;
namespace AntJob
namespace AntJob.Extensions;
/// <summary>SQL语句处理器，定时执行SQL语句</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class SqlHandler : Handler
{
/// <summary>SQL语句处理器，定时执行SQL语句</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class SqlHandler : Handler
#region 构造
/// <summary>实例化</summary>
public SqlHandler()
{
#region 构造
/// <summary>实例化</summary>
public SqlHandler()
//Mode = JobModes.Sql;
var job = Job;
job.BatchSize = 8;
}
#endregion
/// <summary>执行</summary>
/// <param name="ctx"></param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
//var sqls = ctx.Task.Data as String;
var sqls = Job.Data;
sqls = TemplateHelper.Build(sqls, ctx.Task.Start, ctx.Task.End);
// 向调度中心返回解析后的Sql语句
ctx.Remark = sqls;
var sections = SqlSection.ParseAll(sqls);
if (sections.Length == 0) return -1;
var rs = ExecuteSql(sections, ctx, (section, dt) => SqlMessage.ProduceMessage(dt, ctx));
return rs;
}
/// <summary>执行Sql集合</summary>
/// <param name="sections"></param>
/// <param name="ctx"></param>
/// <param name="callback"></param>
/// <returns></returns>
public static Int32 ExecuteSql(SqlSection[] sections, JobContext ctx, Action<SqlSection, DbTable> callback = null)
{
if (sections == null || sections.Length == 0) return -1;
var rs = 0;
ctx.Total = 0;
// 打开事务
foreach (var item in sections)
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).BeginTransaction();
try
{
//Mode = JobModes.Sql;
// 按顺序执行处理Sql语句
DbTable dt = null;
foreach (var section in sections)
switch (section.Action)
{
case SqlActions.Query:
dt = section.Query();
if (dt != null) ctx.Total += dt.Rows.Count;
var job = Job;
job.BatchSize = 8;
}
#endregion
// 处理生产消息
callback?.Invoke(section, dt);
/// <summary>执行</summary>
/// <param name="ctx"></param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
//var sqls = ctx.Task.Data as String;
var sqls = Job.Data;
sqls = TemplateHelper.Build(sqls, ctx.Task.Start, ctx.Task.End);
// 向调度中心返回解析后的Sql语句
ctx.Remark = sqls;
break;
case SqlActions.Execute:
rs += section.Execute();
break;
case SqlActions.Insert:
if (dt.Rows.Count > 0) rs += section.BatchInsert(dt);
break;
default:
break;
}
var sections = SqlSection.ParseAll(sqls);
if (sections.Length == 0) return -1;
var rs = ExecuteSql(sections, ctx, (section, dt) => SqlMessage.ProduceMessage(dt, ctx));
return rs;
}
/// <summary>执行Sql集合</summary>
/// <param name="sections"></param>
/// <param name="ctx"></param>
/// <param name="callback"></param>
/// <returns></returns>
public static Int32 ExecuteSql(SqlSection[] sections, JobContext ctx, Action<SqlSection, DbTable> callback = null)
{
if (sections == null || sections.Length == 0) return -1;
var rs = 0;
ctx.Total = 0;
// 打开事务
// 提交事务
foreach (var item in sections)
{
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).BeginTransaction();
}
try
{
// 按顺序执行处理Sql语句
DbTable dt = null;
foreach (var section in sections)
{
switch (section.Action)
{
case SqlActions.Query:
dt = section.Query();
if (dt != null) ctx.Total += dt.Rows.Count;
// 处理生产消息
callback?.Invoke(section, dt);
break;
case SqlActions.Execute:
rs += section.Execute();
break;
case SqlActions.Insert:
if (dt.Rows.Count > 0) rs += section.BatchInsert(dt);
break;
default:
break;
}
}
// 提交事务
foreach (var item in sections)
{
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).Commit();
}
}
catch
{
// 回滚事务
foreach (var item in sections)
{
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).Rollback();
}
throw;
}
return rs;
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).Commit();
}
catch
{
// 回滚事务
foreach (var item in sections)
if (item.Action != SqlActions.Query) DAL.Create(item.ConnName).Rollback();
throw;
}
return rs;
}
}

102
AntJob.Extensions/SqlMessage.cs
View File

@ -1,67 +1,61 @@
using System;
using System.Linq;
using AntJob.Data;
using AntJob.Extensions;
using AntJob.Data;
using AntJob.Handlers;
using AntJob.Providers;
using NewLife;
using NewLife.Data;
namespace AntJob
namespace AntJob.Extensions;
/// <summary>SQL消息处理器，使用消息匹配SQL语句</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class SqlMessage : MessageHandler
{
/// <summary>SQL消息处理器，使用消息匹配SQL语句</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class SqlMessage : MessageHandler
#region 构造
/// <summary>实例化</summary>
public SqlMessage() => Topic = "Sql";
#endregion
/// <summary>根据解码后的消息执行任务</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
#region 构造
/// <summary>实例化</summary>
public SqlMessage() => Topic = "Sql";
#endregion
var msgs = ctx.Data as String[];
var sqls = Job.Data;
sqls = TemplateHelper.Build(sqls, msgs);
// 向调度中心返回解析后的Sql语句
ctx.Remark = sqls;
/// <summary>根据解码后的消息执行任务</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
// 分解Sql语句得到片段数组
var sections = SqlSection.ParseAll(sqls);
if (sections.Length == 0) return -1;
// 依次执行Sql片段数组。遇到query时，可能需要生产消息
var rs = SqlHandler.ExecuteSql(sections, ctx, (section, dt) => ProduceMessage(dt, ctx));
return rs;
}
/// <summary>根据查询结果生产消息</summary>
/// <param name="dt"></param>
/// <param name="ctx"></param>
public static void ProduceMessage(DbTable dt, JobContext ctx)
{
if (dt == null || dt.Columns == null || dt.Columns.Length == 0 || dt.Rows == null || dt.Rows.Count == 0) return;
// select id as topic_roleId, id as topic_myId from role where updatetime>='{Start}' and updatetime<'{End}'
for (var i = 0; i < dt.Columns.Length; i++)
{
var msgs = ctx.Data as String[];
var sqls = Job.Data;
sqls = TemplateHelper.Build(sqls, msgs);
// 向调度中心返回解析后的Sql语句
ctx.Remark = sqls;
// 分解Sql语句得到片段数组
var sections = SqlSection.ParseAll(sqls);
if (sections.Length == 0) return -1;
// 依次执行Sql片段数组。遇到query时，可能需要生产消息
var rs = SqlHandler.ExecuteSql(sections, ctx, (section, dt) => ProduceMessage(dt, ctx));
return rs;
}
/// <summary>根据查询结果生产消息</summary>
/// <param name="dt"></param>
/// <param name="ctx"></param>
public static void ProduceMessage(DbTable dt, JobContext ctx)
{
if (dt == null || dt.Columns == null || dt.Columns.Length == 0 || dt.Rows == null || dt.Rows.Count == 0) return;
// select id as topic_roleId, id as topic_myId from role where updatetime>='{Start}' and updatetime<'{End}'
for (var i = 0; i < dt.Columns.Length; i++)
var col = dt.Columns[i];
if (col.StartsWithIgnoreCase("topic_"))
{
var col = dt.Columns[i];
if (col.StartsWithIgnoreCase("topic_"))
{
var topic = col.Substring("topic_".Length);
var messages = dt.Rows.Select(e => e[i] + "").Distinct().ToArray();
if (messages.Length > 0)
{
ctx.Handler.Produce(topic, messages, new MessageOption { Unique = true });
}
}
var topic = col.Substring("topic_".Length);
var messages = dt.Rows.Select(e => e[i] + "").Distinct().ToArray();
if (messages.Length > 0)
ctx.Handler.Produce(topic, messages, new MessageOption { Unique = true });
}
}
}

295
AntJob.Extensions/SqlSection.cs
View File

@ -1,166 +1,161 @@
using System;
using System.Collections.Generic;
using System.Linq;
using NewLife;
using NewLife;
using NewLife.Data;
using NewLife.Model;
using NewLife.Reflection;
using XCode.DataAccessLayer;
namespace AntJob.Extensions
namespace AntJob.Extensions;
/// <summary>Sql操作</summary>
public enum SqlActions
{
/// <summary>Sql操作</summary>
public enum SqlActions
/// <summary>查询。返回数据集</summary>
Query,
/// <summary>执行Sql或存储过程。返回影响行数</summary>
Execute,
/// <summary>插入。批量插入数据，需要传入数据集</summary>
Insert,
}
/// <summary>Sql片段。解析sql语句集合</summary>
public class SqlSection
{
#region 属性
/// <summary>连接名</summary>
public String ConnName { get; set; }
/// <summary>操作。添删改查等</summary>
public SqlActions Action { get; set; }
/// <summary>Sql语句</summary>
public String Sql { get; set; }
#endregion
#region 解析
/// <summary>分析sql语句集合，得到片段集合，以双换行分隔</summary>
/// <param name="sqls"></param>
/// <returns></returns>
public static SqlSection[] ParseAll(String sqls)
{
/// <summary>查询。返回数据集</summary>
Query,
var list = new List<SqlSection>();
if (sqls.IsNullOrEmpty()) return list.ToArray();
/// <summary>执行Sql或存储过程。返回影响行数</summary>
Execute,
/// <summary>插入。批量插入数据，需要传入数据集</summary>
Insert,
}
/// <summary>Sql片段。解析sql语句集合</summary>
public class SqlSection
{
#region 属性
/// <summary>连接名</summary>
public String ConnName { get; set; }
/// <summary>操作。添删改查等</summary>
public SqlActions Action { get; set; }
/// <summary>Sql语句</summary>
public String Sql { get; set; }
#endregion
#region 解析
/// <summary>分析sql语句集合，得到片段集合，以双换行分隔</summary>
/// <param name="sqls"></param>
/// <returns></returns>
public static SqlSection[] ParseAll(String sqls)
// 两个换行隔开片段
var ss = sqls.Split(new[] { "\r\n\r", "\r\r", "\n\n" }, StringSplitOptions.RemoveEmptyEntries);
var connName = "";
foreach (var item in ss)
{
var list = new List<SqlSection>();
if (sqls.IsNullOrEmpty()) return list.ToArray();
var section = new SqlSection();
section.Parse(item);
// 两个换行隔开片段
var ss = sqls.Split(new[] { "\r\n\r", "\r\r", "\n\n" }, StringSplitOptions.RemoveEmptyEntries);
var connName = "";
foreach (var item in ss)
{
var section = new SqlSection();
section.Parse(item);
// 如果当前片段未指定连接名，则使用上一个
if (section.ConnName.IsNullOrEmpty())
section.ConnName = connName;
else
connName = section.ConnName;
if (section.ConnName.IsNullOrEmpty()) throw new Exception("未指定连接名！");
list.Add(section);
}
return list.ToArray();
}
/// <summary>分析sql语句到片段</summary>
/// <param name="sql"></param>
public void Parse(String sql)
{
// 解析出来连接名
var str = sql.Substring("/*", "*/")?.Trim();
if (!str.IsNullOrEmpty()) ConnName = str.Substring("use ")?.Trim();
// 剩下的Sql
sql = sql.Substring("*/")?.Trim()?.Trim(';')?.Trim();
Sql = sql;
// 猜测操作
if (sql.StartsWithIgnoreCase("select "))
Action = SqlActions.Query;
else if (sql.StartsWithIgnoreCase("insert into ", "update ", "delete "))
Action = SqlActions.Execute;
// 批量插入
else if (sql.StartsWithIgnoreCase("insert "))
Action = SqlActions.Insert;
// 默认执行，可能是存储过程
// 如果当前片段未指定连接名，则使用上一个
if (section.ConnName.IsNullOrEmpty())
section.ConnName = connName;
else
Action = SqlActions.Execute;
}
#endregion
connName = section.ConnName;
#region 执行处理
/// <summary>查询数据集</summary>
/// <returns></returns>
public DbTable Query() => DAL.Create(ConnName).Query(Sql);
if (section.ConnName.IsNullOrEmpty()) throw new Exception("未指定连接名！");
/// <summary>执行</summary>
/// <returns></returns>
public Int32 Execute()
{
var dal = DAL.Create(ConnName);
// 解析数据表，如果目标表不存在，则返回
var tableName = "";
if (Sql.StartsWithIgnoreCase("delete "))
tableName = Sql.Substring(" from ", " ")?.Trim();
else if (Sql.StartsWithIgnoreCase("udpate "))
tableName = Sql.Substring("udpate ", " ")?.Trim();
if (!tableName.IsNullOrEmpty())
{
var table = dal.Tables?.FirstOrDefault(e => e.TableName.EqualIgnoreCase(tableName));
if (table == null) return -1;
}
return dal.Execute(Sql);
list.Add(section);
}
/// <summary>批量插入</summary>
/// <param name="dt"></param>
/// <returns></returns>
public Int32 BatchInsert(DbTable dt)
{
var tableName = Sql.Substring(" ")?.Trim(';');
var dal = DAL.Create(ConnName);
// 执行反向工程，该建表就建表
//dal.CheckDatabase();
var table = dal.Tables?.FirstOrDefault(e => e.TableName.EqualIgnoreCase(tableName));
//if (table == null) throw new Exception($"在连接[{ConnName}]中无法找到数据表[{tableName}]");
if (table == null)
{
var ioc = ObjectContainer.Current;
table = ioc.Resolve<IDataTable>();
table.TableName = tableName;
for (var i = 0; i < dt.Columns.Length; i++)
{
var dc = table.CreateColumn();
dc.ColumnName = dt.Columns[i];
dc.DataType = dt.Types[i];
table.Columns.Add(dc);
}
dal.SetTables(table);
}
// 选取目标表和数据集共有的字段
tableName = dal.Db.FormatName(table);
var columns = new List<IDataColumn>();
foreach (var dc in table.Columns)
{
if (dc.ColumnName.EqualIgnoreCase(dt.Columns)) columns.Add(dc);
}
return dal.Session.Insert(table, columns.ToArray(), dt.Cast<IModel>());
}
#endregion
return list.ToArray();
}
/// <summary>分析sql语句到片段</summary>
/// <param name="sql"></param>
public void Parse(String sql)
{
// 解析出来连接名
var str = sql.Substring("/*", "*/")?.Trim();
if (!str.IsNullOrEmpty()) ConnName = str.Substring("use ")?.Trim();
// 剩下的Sql
sql = sql.Substring("*/")?.Trim()?.Trim(';')?.Trim();
Sql = sql;
// 猜测操作
if (sql.StartsWithIgnoreCase("select "))
Action = SqlActions.Query;
else if (sql.StartsWithIgnoreCase("insert into ", "update ", "delete "))
Action = SqlActions.Execute;
// 批量插入
else if (sql.StartsWithIgnoreCase("insert "))
Action = SqlActions.Insert;
// 默认执行，可能是存储过程
else
Action = SqlActions.Execute;
}
#endregion
#region 执行处理
/// <summary>查询数据集</summary>
/// <returns></returns>
public DbTable Query() => DAL.Create(ConnName).Query(Sql);
/// <summary>执行</summary>
/// <returns></returns>
public Int32 Execute()
{
var dal = DAL.Create(ConnName);
// 解析数据表，如果目标表不存在，则返回
var tableName = "";
if (Sql.StartsWithIgnoreCase("delete "))
tableName = Sql.Substring(" from ", " ")?.Trim();
else if (Sql.StartsWithIgnoreCase("udpate "))
tableName = Sql.Substring("udpate ", " ")?.Trim();
if (!tableName.IsNullOrEmpty())
{
var table = dal.Tables?.FirstOrDefault(e => e.TableName.EqualIgnoreCase(tableName));
if (table == null) return -1;
}
return dal.Execute(Sql);
}
/// <summary>批量插入</summary>
/// <param name="dt"></param>
/// <returns></returns>
public Int32 BatchInsert(DbTable dt)
{
var tableName = Sql.Substring(" ")?.Trim(';');
var dal = DAL.Create(ConnName);
// 执行反向工程，该建表就建表
//dal.CheckDatabase();
var table = dal.Tables?.FirstOrDefault(e => e.TableName.EqualIgnoreCase(tableName));
//if (table == null) throw new Exception($"在连接[{ConnName}]中无法找到数据表[{tableName}]");
if (table == null)
{
var ioc = ObjectContainer.Current;
table = ioc.Resolve<IDataTable>();
table.TableName = tableName;
for (var i = 0; i < dt.Columns.Length; i++)
{
var dc = table.CreateColumn();
dc.ColumnName = dt.Columns[i];
dc.DataType = dt.Types[i];
table.Columns.Add(dc);
}
dal.SetTables(table);
}
// 选取目标表和数据集共有的字段
tableName = dal.Db.FormatName(table);
var columns = new List<IDataColumn>();
foreach (var dc in table.Columns)
{
if (dc.ColumnName.EqualIgnoreCase(dt.Columns)) columns.Add(dc);
}
return dal.Session.Insert(table, columns.ToArray(), dt.Cast<IModel>());
}
#endregion
}

3
AntJob.Server/Worker.cs
View File

@ -7,6 +7,7 @@ using NewLife.Log;
using NewLife.Remoting;
using NewLife.Threading;
using Stardust.Registry;
using XCode;
namespace AntJob.Server;
@ -75,7 +76,7 @@ public class Worker : IHostedService
set.Save();
}
var set2 = XCode.Setting.Current;
var set2 = XCodeSetting.Current;
if (set2.IsNew)
{
set2.Debug = true;

4
AntJob/AntJob.csproj
View File

@ -1,12 +1,12 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<TargetFrameworks>netstandard2.1;netstandard2.0;net461</TargetFrameworks>
<TargetFrameworks>netstandard2.1;netstandard2.0;net461;net45</TargetFrameworks>
<AssemblyTitle>蚂蚁调度SDK</AssemblyTitle>
<Description>分布式任务调度系统，纯NET打造的重量级大数据实时计算平台，万亿级调度经验积累。</Description>
<Company>新生命开发团队</Company>
<Copyright>©2002-2023 NewLife</Copyright>
<VersionPrefix>3.2</VersionPrefix>
<VersionPrefix>3.3</VersionPrefix>
<VersionSuffix>$([System.DateTime]::Now.ToString(`yyyy.MMdd`))</VersionSuffix>
<Version>$(VersionPrefix).$(VersionSuffix)</Version>
<FileVersion>$(Version)</FileVersion>

68
AntJob/AntSetting.cs
View File

@ -1,45 +1,43 @@
using System;
using System.ComponentModel;
using System.ComponentModel;
using System.Reflection;
using NewLife;
using NewLife.Configuration;
namespace AntJob
namespace AntJob;
/// <summary>蚂蚁配置。主要用于网络型调度系统</summary>
[Config("Ant")]
public class AntSetting : Config<AntSetting>
{
/// <summary>蚂蚁配置。主要用于网络型调度系统</summary>
[Config("Ant")]
public class AntSetting : Config<AntSetting>
#region 属性
/// <summary>调试开关。默认false</summary>
[Description("调试开关。默认false")]
public Boolean Debug { get; set; }
/// <summary>调度中心。逗号分隔多地址，主备架构</summary>
[Description("调度中心。逗号分隔多地址，主备架构")]
public String Server { get; set; } = "tcp://127.0.0.1:9999,tcp://ant.newlifex.com:9999";
/// <summary>应用标识。调度中心以此隔离应用，默认当前应用</summary>
[Description("应用标识。调度中心以此隔离应用，默认当前应用")]
public String AppID { get; set; }
/// <summary>应用密钥。</summary>
[Description("应用密钥。")]
public String Secret { get; set; }
#endregion
#region 方法
/// <summary>重载</summary>
protected override void OnLoaded()
{
#region 属性
/// <summary>调试开关。默认false</summary>
[Description("调试开关。默认false")]
public Boolean Debug { get; set; }
/// <summary>调度中心。逗号分隔多地址，主备架构</summary>
[Description("调度中心。逗号分隔多地址，主备架构")]
public String Server { get; set; } = "tcp://127.0.0.1:9999,tcp://ant.newlifex.com:9999";
/// <summary>应用标识。调度中心以此隔离应用，默认当前应用</summary>
[Description("应用标识。调度中心以此隔离应用，默认当前应用")]
public String AppID { get; set; }
/// <summary>应用密钥。</summary>
[Description("应用密钥。")]
public String Secret { get; set; }
#endregion
#region 方法
/// <summary>重载</summary>
protected override void OnLoaded()
if (AppID.IsNullOrEmpty())
{
if (AppID.IsNullOrEmpty())
{
var asm = Assembly.GetEntryAssembly();
if (asm != null) AppID = asm.GetName().Name;
}
base.OnLoaded();
var asm = Assembly.GetEntryAssembly();
if (asm != null) AppID = asm.GetName().Name;
}
#endregion
base.OnLoaded();
}
#endregion
}

186
AntJob/Data/TemplateHelper.cs
View File

@ -1,114 +1,112 @@
using System;
using NewLife;
using NewLife;
using NewLife.Collections;
namespace AntJob.Data
namespace AntJob.Data;
/// <summary>模板助手</summary>
public static class TemplateHelper
{
/// <summary>模板助手</summary>
public static class TemplateHelper
/// <summary>使用时间参数处理模板</summary>
/// <param name="template"></param>
/// <param name="startTime"></param>
/// <param name="endTime"></param>
/// <returns></returns>
public static String Build(String template, DateTime startTime, DateTime endTime)
{
/// <summary>使用时间参数处理模板</summary>
/// <param name="template"></param>
/// <param name="startTime"></param>
/// <param name="endTime"></param>
/// <returns></returns>
public static String Build(String template, DateTime startTime, DateTime endTime)
if (template.IsNullOrEmpty()) return template;
var str = template;
var sb = Pool.StringBuilder.Get();
var p = 0;
while (true)
{
if (template.IsNullOrEmpty()) return template;
var str = template;
var sb = Pool.StringBuilder.Get();
var p = 0;
while (true)
var ti = Find(str, "Start", p);
if (ti == null)
{
var ti = Find(str, "Start", p);
if (ti == null)
{
sb.Append(str.Substring(p));
break;
}
// 准备替换
var val = ti.Item3.IsNullOrEmpty() ? startTime.ToFullString() : startTime.ToString(ti.Item3);
sb.Append(str.Substring(p, ti.Item1 - p));
sb.Append(val);
// 移动指针
p = ti.Item2 + 1;
sb.Append(str.Substring(p));
break;
}
str = sb.ToString();
sb.Clear();
p = 0;
while (true)
// 准备替换
var val = ti.Item3.IsNullOrEmpty() ? startTime.ToFullString() : startTime.ToString(ti.Item3);
sb.Append(str.Substring(p, ti.Item1 - p));
sb.Append(val);
// 移动指针
p = ti.Item2 + 1;
}
str = sb.ToString();
sb.Clear();
p = 0;
while (true)
{
var ti = Find(str, "End", p);
if (ti == null)
{
var ti = Find(str, "End", p);
if (ti == null)
{
sb.Append(str.Substring(p));
break;
}
// 准备替换
var val = ti.Item3.IsNullOrEmpty() ? endTime.ToFullString() : endTime.ToString(ti.Item3);
sb.Append(str.Substring(p, ti.Item1 - p));
sb.Append(val);
// 移动指针
p = ti.Item2 + 1;
sb.Append(str.Substring(p));
break;
}
return sb.Put(true);
// 准备替换
var val = ti.Item3.IsNullOrEmpty() ? endTime.ToFullString() : endTime.ToString(ti.Item3);
sb.Append(str.Substring(p, ti.Item1 - p));
sb.Append(val);
// 移动指针
p = ti.Item2 + 1;
}
private static Tuple<Int32, Int32, String> Find(String str, String key, Int32 p)
return sb.Put(true);
}
private static Tuple<Int32, Int32, String> Find(String str, String key, Int32 p)
{
// 头尾
var p1 = str.IndexOf("{" + key, p);
if (p1 < 0) return null;
var p2 = str.IndexOf("}", p1);
if (p2 < 0) return null;
// 格式化字符串
var format = "";
var p3 = str.IndexOf(":", p1);
if (p3 > 0 && p3 < p2) format = str.Substring(p3 + 1, p2 - p3 - 1);
// 左括号位置，右括号位置，格式化字符串
return new Tuple<Int32, Int32, String>(p1, p2, format);
}
/// <summary>使用消息数组处理模板</summary>
/// <param name="template"></param>
/// <param name="messages"></param>
/// <returns></returns>
public static String Build(String template, String[] messages)
{
if (template.IsNullOrEmpty()) return template;
var str = template;
var sb = Pool.StringBuilder.Get();
var p = 0;
while (true)
{
// 头尾
var p1 = str.IndexOf("{" + key, p);
if (p1 < 0) return null;
var p2 = str.IndexOf("}", p1);
if (p2 < 0) return null;
// 格式化字符串
var format = "";
var p3 = str.IndexOf(":", p1);
if (p3 > 0 && p3 < p2) format = str.Substring(p3 + 1, p2 - p3 - 1);
// 左括号位置，右括号位置，格式化字符串
return new Tuple<Int32, Int32, String>(p1, p2, format);
}
/// <summary>使用消息数组处理模板</summary>
/// <param name="template"></param>
/// <param name="messages"></param>
/// <returns></returns>
public static String Build(String template, String[] messages)
{
if (template.IsNullOrEmpty()) return template;
var str = template;
var sb = Pool.StringBuilder.Get();
var p = 0;
while (true)
var p1 = str.IndexOf("{Message}", p);
if (p1 < 0)
{
var p1 = str.IndexOf("{Message}", p);
if (p1 < 0)
{
sb.Append(str.Substring(p));
break;
}
// 准备替换
var val = messages.Join();
sb.Append(str.Substring(p, p1 - p));
sb.Append(val);
// 移动指针
p = p1 + "{Message}".Length;
sb.Append(str.Substring(p));
break;
}
return sb.Put(true);
// 准备替换
var val = messages.Join();
sb.Append(str.Substring(p, p1 - p));
sb.Append(val);
// 移动指针
p = p1 + "{Message}".Length;
}
return sb.Put(true);
}
}

395
AntJob/Handler.cs
View File

@ -1,211 +1,212 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Diagnostics;
using AntJob.Data;
using AntJob.Providers;
using NewLife;
using NewLife.Collections;
using NewLife.Log;
namespace AntJob
namespace AntJob;
/// <summary>处理器基类，每个作业一个处理器</summary>
/// <remarks>
/// 每个作业一个处理器类，负责一个业务处理模块。
/// 例如在数据同步或数据清洗中，每张表就写一个处理器，如果一组数据表有共同特性，还可以为它们封装一个自己的处理器基类。
///
/// 定时调度只要当前时间达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// </remarks>
public abstract class Handler
{
/// <summary>处理器基类，每个作业一个处理器</summary>
/// <remarks>
/// 每个作业一个处理器类，负责一个业务处理模块。
/// 例如在数据同步或数据清洗中，每张表就写一个处理器，如果一组数据表有共同特性，还可以为它们封装一个自己的处理器基类。
///
/// 定时调度只要达到时间片开头就可以跑，数据调度要求达到时间片末尾才可以跑。
/// </remarks>
public abstract class Handler
#region 属性
/// <summary>名称</summary>
public String Name { get; set; }
/// <summary>调度器</summary>
public Scheduler Schedule { get; set; }
/// <summary>作业提供者</summary>
public IJobProvider Provider { get; set; }
/// <summary>作业模型。启动前作为创建作业的默认值，启动后表示作业当前设置和状态</summary>
public IJob Job { get; set; }
/// <summary>是否工作中</summary>
public Boolean Active { get; private set; }
/// <summary>调度模式</summary>
public virtual JobModes Mode { get; set; } = JobModes.Alarm;
private volatile Int32 _Busy;
/// <summary>正在处理中的任务数</summary>
public Int32 Busy => _Busy;
/// <summary>性能跟踪器</summary>
public ITracer Tracer { get; set; }
#endregion
#region 索引器
private readonly IDictionary<String, Object> _Items = new NullableDictionary<String, Object>(StringComparer.OrdinalIgnoreCase);
/// <summary>用户数据</summary>
/// <param name="item"></param>
/// <returns></returns>
public Object this[String item] { get => _Items[item]; set => _Items[item] = value; }
#endregion
#region 构造
/// <summary>实例化</summary>
public Handler()
{
#region 属性
/// <summary>名称</summary>
public String Name { get; set; }
Name = GetType().Name.TrimEnd(nameof(Handler));
/// <summary>调度器</summary>
public Scheduler Schedule { get; set; }
/// <summary>作业提供者</summary>
public IJobProvider Provider { get; set; }
/// <summary>作业模型。启动前作为创建作业的默认值，启动后表示作业当前设置和状态</summary>
public IJob Job { get; set; }
/// <summary>是否工作中</summary>
public Boolean Active { get; private set; }
/// <summary>调度模式</summary>
public virtual JobModes Mode { get; set; } = JobModes.Alarm;
private volatile Int32 _Busy;
/// <summary>正在处理中的任务数</summary>
public Int32 Busy => _Busy;
#endregion
#region 索引器
private readonly IDictionary<String, Object> _Items = new NullableDictionary<String, Object>(StringComparer.OrdinalIgnoreCase);
/// <summary>用户数据</summary>
/// <param name="item"></param>
/// <returns></returns>
public Object this[String item] { get => _Items[item]; set => _Items[item] = value; }
#endregion
#region 构造
/// <summary>实例化</summary>
public Handler()
// 默认本月1号
var now = DateTime.Now;
var job = new JobModel
{
Name = GetType().Name.TrimEnd(nameof(Handler));
Start = new DateTime(now.Year, now.Month, 1),
Step = 30,
Offset = 15,
Mode = JobModes.Alarm,
};
// 默认本月1号
var now = DateTime.Now;
var job = new JobModel
{
Start = new DateTime(now.Year, now.Month, 1),
Step = 30,
Offset = 15,
Mode = JobModes.Alarm,
};
// 默认并发数为核心数
job.MaxTask = Environment.ProcessorCount;
if (job.MaxTask < 8) job.MaxTask = 8;
// 默认并发数为核心数
job.MaxTask = Environment.ProcessorCount;
if (job.MaxTask > 8) job.MaxTask = 8;
Job = job;
}
#endregion
#region 基本方法
/// <summary>开始</summary>
public virtual Boolean Start()
{
if (Active) return false;
var msg = "开始工作";
var job = Job;
if (job != null) msg += $" {job.Enable} 区间（{job.Start}, {job.End}） Offset={job.Offset} Step={job.Step} MaxTask={job.MaxTask}";
WriteLog(msg);
Active = true;
return true;
}
/// <summary>停止</summary>
public virtual Boolean Stop()
{
if (!Active) return false;
WriteLog("停止工作");
Active = false;
return true;
}
#endregion
#region 申请任务
/// <summary>申请任务</summary>
/// <remarks>
/// 业务应用根据使用场景，可重载Acquire并返回空来阻止创建新任务
/// </remarks>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public virtual ITask[] Acquire(Int32 count)
{
var prv = Provider;
var job = Job;
// 循环申请任务，喂饱处理器
return prv.Acquire(job, null, count);
}
#endregion
#region 整体调度
/// <summary>准备就绪，增加Busy，避免超额分配</summary>
/// <param name="task"></param>
internal void Prepare(ITask task) => Interlocked.Increment(ref _Busy);
/// <summary>处理一项新任务</summary>
/// <param name="task"></param>
public void Process(ITask task)
{
if (task == null) return;
var ctx = new JobContext
{
Handler = this,
Task = task,
Result = new TaskResult { ID = task.ID },
};
// APM埋点
var span = Schedule.Tracer?.NewSpan($"job:{Name}", task.Data ?? $"({task.Start.ToFullString()}, {task.End.ToFullString()})");
ctx.Remark = span?.ToString();
var sw = Stopwatch.StartNew();
try
{
OnProcess(ctx);
}
catch (Exception ex)
{
ctx.Error = ex;
span?.SetError(ex, task);
XTrace.WriteException(ex);
}
finally
{
Interlocked.Decrement(ref _Busy);
span?.Dispose();
}
sw.Stop();
ctx.Cost = sw.Elapsed.TotalMilliseconds;
OnFinish(ctx);
Schedule?.OnFinish(ctx);
ctx.Items.Clear();
}
/// <summary>处理任务。内部分批处理</summary>
/// <param name="ctx"></param>
protected virtual void OnProcess(JobContext ctx)
{
ctx.Total = 1;
ctx.Success = Execute(ctx);
}
#endregion
#region 数据处理
/// <summary>处理一批数据，一个任务内多次调用</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected abstract Int32 Execute(JobContext ctx);
/// <summary>生产消息</summary>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public Int32 Produce(String topic, String[] messages, MessageOption option = null) => Provider.Produce(Job?.Name, topic, messages, option);
/// <summary>整个任务完成</summary>
/// <param name="ctx"></param>
protected virtual void OnFinish(JobContext ctx) => Provider?.Finish(ctx);
#endregion
#region 日志
/// <summary>日志</summary>
public ILog Log { get; set; } = Logger.Null;
/// <summary>写日志</summary>
/// <param name="format"></param>
/// <param name="args"></param>
public void WriteLog(String format, params Object[] args) => Log?.Info(Name + " " + format, args);
#endregion
Job = job;
}
#endregion
#region 基本方法
/// <summary>开始</summary>
public virtual Boolean Start()
{
if (Active) return false;
var msg = "开始工作";
var job = Job;
if (job != null) msg += $" {job.Enable} 区间（{job.Start}, {job.End}） Offset={job.Offset} Step={job.Step} MaxTask={job.MaxTask}";
using var span = Tracer?.NewSpan($"job:{Name}:Start", msg);
WriteLog(msg);
Active = true;
return true;
}
/// <summary>停止</summary>
public virtual Boolean Stop(String reason)
{
if (!Active) return false;
using var span = Tracer?.NewSpan($"job:{Name}:Stop", reason);
WriteLog("停止工作");
Active = false;
return true;
}
#endregion
#region 申请任务
/// <summary>申请任务</summary>
/// <remarks>
/// 业务应用根据使用场景，可重载Acquire并返回空来阻止创建新任务
/// </remarks>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public virtual ITask[] Acquire(Int32 count)
{
var prv = Provider;
var job = Job;
// 循环申请任务，喂饱处理器
return prv.Acquire(job, null, count);
}
#endregion
#region 整体调度
/// <summary>准备就绪，增加Busy，避免超额分配</summary>
/// <param name="task"></param>
internal void Prepare(ITask task) => Interlocked.Increment(ref _Busy);
/// <summary>处理一项新任务</summary>
/// <param name="task"></param>
public void Process(ITask task)
{
if (task == null) return;
var ctx = new JobContext
{
Handler = this,
Task = task,
Result = new TaskResult { ID = task.ID },
};
// APM埋点
var span = Schedule.Tracer?.NewSpan($"job:{Name}", task.Data ?? $"({task.Start.ToFullString()}, {task.End.ToFullString()})");
ctx.Remark = span?.ToString();
var sw = Stopwatch.StartNew();
try
{
OnProcess(ctx);
}
catch (Exception ex)
{
ctx.Error = ex;
span?.SetError(ex, task);
XTrace.WriteException(ex);
}
finally
{
Interlocked.Decrement(ref _Busy);
span?.Dispose();
}
sw.Stop();
ctx.Cost = sw.Elapsed.TotalMilliseconds;
OnFinish(ctx);
Schedule?.OnFinish(ctx);
ctx.Items.Clear();
}
/// <summary>处理任务。内部分批处理</summary>
/// <param name="ctx"></param>
protected virtual void OnProcess(JobContext ctx)
{
ctx.Total = 1;
ctx.Success = Execute(ctx);
}
#endregion
#region 数据处理
/// <summary>处理一批数据，一个任务内多次调用</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected abstract Int32 Execute(JobContext ctx);
/// <summary>生产消息</summary>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public Int32 Produce(String topic, String[] messages, MessageOption option = null) => Provider.Produce(Job?.Name, topic, messages, option);
/// <summary>整个任务完成</summary>
/// <param name="ctx"></param>
protected virtual void OnFinish(JobContext ctx) => Provider?.Finish(ctx);
#endregion
#region 日志
/// <summary>日志</summary>
public ILog Log { get; set; } = Logger.Null;
/// <summary>写日志</summary>
/// <param name="format"></param>
/// <param name="args"></param>
public void WriteLog(String format, params Object[] args) => Log?.Info(Name + " " + format, args);
#endregion
}

61
AntJob/Handlers/CSharpHandler.cs
View File

@ -1,42 +1,35 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using AntJob.Data;
using NewLife;
using NewLife;
namespace AntJob.Handlers
namespace AntJob.Handlers;
/// <summary>C#代码处理器，定时执行一段C#代码</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class CSharpHandler : Handler
{
/// <summary>C#代码处理器，定时执行一段C#代码</summary>
/// <remarks>
/// 应用型处理器，可直接使用
/// </remarks>
public class CSharpHandler : Handler
#region 属性
#endregion
#region 构造
/// <summary>实例化</summary>
public CSharpHandler()
{
#region 属性
#endregion
//Mode = JobModes.CSharp;
#region 构造
/// <summary>实例化</summary>
public CSharpHandler()
{
//Mode = JobModes.CSharp;
var job = Job;
job.BatchSize = 8;
}
#endregion
var job = Job;
job.BatchSize = 8;
}
#endregion
/// <summary>执行</summary>
/// <param name="ctx"></param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var code = ctx.Data as String;
if (code.IsNullOrWhiteSpace()) return -1;
/// <summary>执行</summary>
/// <param name="ctx"></param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var code = ctx.Data as String;
if (code.IsNullOrWhiteSpace()) return -1;
return 0;
}
return 0;
}
}

186
AntJob/Handlers/MessageHandler.cs
View File

@ -1,104 +1,100 @@
using System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Collections;
using AntJob.Data;
using NewLife;
using NewLife.Serialization;
namespace AntJob.Handlers
namespace AntJob.Handlers;
/// <summary>消息调度基类，消费的消息在Data中返回</summary>
public abstract class MessageHandler : Handler
{
/// <summary>消息调度基类，消费的消息在Data中返回</summary>
public abstract class MessageHandler : Handler
#region 属性
/// <summary>主题。设置后使用消费调度模式</summary>
public String Topic { get; set; }
#endregion
#region 构造
/// <summary>实例化</summary>
public MessageHandler()
{
#region 属性
/// <summary>主题。设置后使用消费调度模式</summary>
public String Topic { get; set; }
#endregion
Mode = JobModes.Message;
#region 构造
/// <summary>实例化</summary>
public MessageHandler()
{
Mode = JobModes.Message;
var job = Job;
job.BatchSize = 8;
}
#endregion
#region 方法
/// <summary>参数检查</summary>
/// <returns></returns>
public override Boolean Start()
{
if (Topic.IsNullOrEmpty()) throw new ArgumentNullException(nameof(Topic), "消息调度要求设置主题");
return base.Start();
}
/// <summary>申请任务</summary>
/// <remarks>
/// 业务应用根据使用场景，可重载Acquire并返回空来阻止创建新任务
/// </remarks>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(Int32 count)
{
// 消费模式，设置Topic值
var prv = Provider;
var job = Job;
return prv.Acquire(job, Topic, count);
}
/// <summary>解码一批消息，处理任务</summary>
/// <param name="ctx"></param>
protected override void OnProcess(JobContext ctx)
{
if (ctx.Task.Data.IsNullOrEmpty()) return;
var ss = ctx.Task.Data.ToJsonEntity<String[]>();
if (ss == null || ss.Length == 0) return;
//// 消息作业特殊优待字符串，不需要再次Json解码
//if (typeof(TModel) == typeof(String))
//{
ctx.Total = ss.Length;
ctx.Data = ss;
//}
//else
//{
// var ms = ss.Select(e => e.ToJsonEntity<TModel>()).ToList();
// ctx.Total = ms.Count;
// ctx.Data = ms;
//}
Execute(ctx);
}
/// <summary>根据解码后的消息执行任务</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var count = 0;
foreach (String item in ctx.Data as IEnumerable)
{
//ctx.Key = item as String;
//ctx.Entity = item;
if (ProcessItem(ctx, item)) count++;
}
return count;
}
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="message">消息</param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, String message) => true;
#endregion
var job = Job;
job.BatchSize = 8;
}
#endregion
#region 方法
/// <summary>参数检查</summary>
/// <returns></returns>
public override Boolean Start()
{
if (Topic.IsNullOrEmpty()) throw new ArgumentNullException(nameof(Topic), "消息调度要求设置主题");
return base.Start();
}
/// <summary>申请任务</summary>
/// <remarks>
/// 业务应用根据使用场景，可重载Acquire并返回空来阻止创建新任务
/// </remarks>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(Int32 count)
{
// 消费模式，设置Topic值
var prv = Provider;
var job = Job;
return prv.Acquire(job, Topic, count);
}
/// <summary>解码一批消息，处理任务</summary>
/// <param name="ctx"></param>
protected override void OnProcess(JobContext ctx)
{
if (ctx.Task.Data.IsNullOrEmpty()) return;
var ss = ctx.Task.Data.ToJsonEntity<String[]>();
if (ss == null || ss.Length == 0) return;
//// 消息作业特殊优待字符串，不需要再次Json解码
//if (typeof(TModel) == typeof(String))
//{
ctx.Total = ss.Length;
ctx.Data = ss;
//}
//else
//{
// var ms = ss.Select(e => e.ToJsonEntity<TModel>()).ToList();
// ctx.Total = ms.Count;
// ctx.Data = ms;
//}
Execute(ctx);
}
/// <summary>根据解码后的消息执行任务</summary>
/// <param name="ctx">上下文</param>
/// <returns></returns>
protected override Int32 Execute(JobContext ctx)
{
var count = 0;
foreach (String item in ctx.Data as IEnumerable)
{
//ctx.Key = item as String;
//ctx.Entity = item;
if (ProcessItem(ctx, item)) count++;
}
return count;
}
/// <summary>处理一个数据对象</summary>
/// <param name="ctx">上下文</param>
/// <param name="message">消息</param>
/// <returns></returns>
protected virtual Boolean ProcessItem(JobContext ctx, String message) => true;
#endregion
}

89
AntJob/JobContext.cs
View File

@ -1,65 +1,62 @@
using System;
using System.Collections.Generic;
using AntJob.Data;
using AntJob.Data;
using NewLife.Collections;
using NewLife.Data;
namespace AntJob
namespace AntJob;
/// <summary>作业上下文</summary>
public class JobContext : IExtend
{
/// <summary>作业上下文</summary>
public class JobContext : IExtend
{
#region 属性
/// <summary>作业</summary>
public Handler Handler { get; set; }
#region 属性
/// <summary>作业</summary>
public Handler Handler { get; set; }
/// <summary>任务参数</summary>
public ITask Task { get; set; }
/// <summary>任务参数</summary>
public ITask Task { get; set; }
/// <summary>任务结果</summary>
public ITaskResult Result { get; set; }
/// <summary>任务结果</summary>
public ITaskResult Result { get; set; }
/// <summary>状态</summary>
public JobStatus Status { get; set; }
/// <summary>状态</summary>
public JobStatus Status { get; set; }
/// <summary>列表数据</summary>
public Object Data { get; set; }
/// <summary>列表数据</summary>
public Object Data { get; set; }
/// <summary>处理总数</summary>
public Int32 Total { get; set; }
/// <summary>处理总数</summary>
public Int32 Total { get; set; }
/// <summary>成功处理数</summary>
public Int32 Success { get; set; }
/// <summary>成功处理数</summary>
public Int32 Success { get; set; }
/// <summary>总耗时，毫秒</summary>
public Double Cost { get; set; }
/// <summary>总耗时，毫秒</summary>
public Double Cost { get; set; }
/// <summary>最后处理键值。由业务决定，便于分析问题</summary>
public String Key { get; set; }
/// <summary>最后处理键值。由业务决定，便于分析问题</summary>
public String Key { get; set; }
///// <summary>当前处理对象</summary>
//public Object Entity { get; set; }
///// <summary>当前处理对象</summary>
//public Object Entity { get; set; }
/// <summary>处理异常</summary>
public Exception Error { get; set; }
/// <summary>处理异常</summary>
public Exception Error { get; set; }
/// <summary>任务备注消息。可用于保存到任务项内容字段</summary>
public String Remark { get; set; }
#endregion
/// <summary>任务备注消息。可用于保存到任务项内容字段</summary>
public String Remark { get; set; }
#endregion
#region 索引器
/// <summary>用户数据</summary>
public IDictionary<String, Object> Items { get; set; } = new NullableDictionary<String, Object>(StringComparer.OrdinalIgnoreCase);
#region 索引器
/// <summary>用户数据</summary>
public IDictionary<String, Object> Items { get; set; } = new NullableDictionary<String, Object>(StringComparer.OrdinalIgnoreCase);
/// <summary>用户数据</summary>
/// <param name="item"></param>
/// <returns></returns>
public Object this[String item] { get => Items[item]; set => Items[item] = value; }
#endregion
/// <summary>用户数据</summary>
/// <param name="item"></param>
/// <returns></returns>
public Object this[String item] { get => Items[item]; set => Items[item] = value; }
#endregion
#region 扩展属性
/// <summary>处理速度</summary>
public Int32 Speed => (Cost <= 0 || Total == 0) ? 0 : (Int32)Math.Min(Total * 1000L / Cost, Int32.MaxValue);
#endregion
}
#region 扩展属性
/// <summary>处理速度</summary>
public Int32 Speed => (Cost <= 0 || Total == 0) ? 0 : (Int32)Math.Min(Total * 1000L / Cost, Int32.MaxValue);
#endregion
}

25
AntJob/Models/AcquireModel.cs
View File

@ -1,21 +1,14 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace AntJob.Models;
namespace AntJob.Models
/// <summary>申请作业任务</summary>
public class AcquireModel
{
/// <summary>申请作业任务</summary>
public class AcquireModel
{
/// <summary>作业名</summary>
public String Job { get; set; }
/// <summary>作业名</summary>
public String Job { get; set; }
/// <summary>主题</summary>
public String Topic { get; set; }
/// <summary>主题</summary>
public String Topic { get; set; }
/// <summary>任务数</summary>
public Int32 Count { get; set; }
}
/// <summary>任务数</summary>
public Int32 Count { get; set; }
}

61
AntJob/Models/LoginModel.cs
View File

@ -1,46 +1,39 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace AntJob.Models;
namespace AntJob.Models
/// <summary>登录模型</summary>
public class LoginModel
{
/// <summary>登录模型</summary>
public class LoginModel
{
/// <summary>用户名</summary>
public String User { get; set; }
/// <summary>用户名</summary>
public String User { get; set; }
/// <summary>用户名</summary>
public String Pass { get; set; }
/// <summary>用户名</summary>
public String Pass { get; set; }
/// <summary>显示名</summary>
public String DisplayName { get; set; }
/// <summary>显示名</summary>
public String DisplayName { get; set; }
/// <summary>机器名</summary>
public String Machine { get; set; }
/// <summary>机器名</summary>
public String Machine { get; set; }
/// <summary>进程Id</summary>
public Int32 ProcessId { get; set; }
/// <summary>进程Id</summary>
public Int32 ProcessId { get; set; }
/// <summary>版本</summary>
public String Version { get; set; }
/// <summary>版本</summary>
public String Version { get; set; }
/// <summary>编译时间</summary>
public DateTime Compile { get; set; }
}
/// <summary>编译时间</summary>
public DateTime Compile { get; set; }
}
/// <summary>登录响应</summary>
public class LoginResponse
{
/// <summary>名称</summary>
public String Name { get; set; }
/// <summary>登录响应</summary>
public class LoginResponse
{
/// <summary>名称</summary>
public String Name { get; set; }
/// <summary>密钥。仅注册时返回</summary>
public String Secret { get; set; }
/// <summary>密钥。仅注册时返回</summary>
public String Secret { get; set; }
/// <summary>显示名</summary>
public String DisplayName { get; set; }
}
/// <summary>显示名</summary>
public String DisplayName { get; set; }
}

33
AntJob/Models/ProduceModel.cs
View File

@ -1,27 +1,20 @@
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace AntJob.Models;
namespace AntJob.Models
/// <summary>生成消息的模型</summary>
public class ProduceModel
{
/// <summary>生成消息的模型</summary>
public class ProduceModel
{
/// <summary>作业名</summary>
public String Job { get; set; }
/// <summary>作业名</summary>
public String Job { get; set; }
/// <summary>主题</summary>
public String Topic { get; set; }
/// <summary>主题</summary>
public String Topic { get; set; }
/// <summary>消息集合</summary>
public String[] Messages { get; set; }
/// <summary>消息集合</summary>
public String[] Messages { get; set; }
/// <summary>延迟执行间隔。（实际执行时间=延迟+生产时间），单位秒</summary>
public Int32 DelayTime { get; set; }
/// <summary>延迟执行间隔。（实际执行时间=延迟+生产时间），单位秒</summary>
public Int32 DelayTime { get; set; }
/// <summary>消息去重。避免单个消息被重复生产</summary>
public Boolean Unique { get; set; }
}
/// <summary>消息去重。避免单个消息被重复生产</summary>
public Boolean Unique { get; set; }
}

289
AntJob/Providers/AntClient.cs
View File

@ -1,10 +1,6 @@
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.ComponentModel;
using System.Diagnostics;
using System.Linq;
using System.Reflection;
using System.Threading.Tasks;
using AntJob.Data;
using AntJob.Models;
using NewLife;
@ -14,153 +10,152 @@ using NewLife.Reflection;
using NewLife.Remoting;
using NewLife.Serialization;
namespace AntJob.Providers
namespace AntJob.Providers;
/// <summary>蚂蚁客户端</summary>
public class AntClient : ApiClient
{
/// <summary>蚂蚁客户端</summary>
public class AntClient : ApiClient
#region 属性
/// <summary>用户名</summary>
public String UserName { get; set; }
/// <summary>密码</summary>
public String Password { get; set; }
/// <summary>是否已登录</summary>
public Boolean Logined { get; set; }
/// <summary>最后一次登录成功后的消息</summary>
public LoginResponse Info { get; private set; }
#endregion
#region 方法
/// <summary>实例化</summary>
public AntClient()
{
#region 属性
/// <summary>用户名</summary>
public String UserName { get; set; }
Log = XTrace.Log;
/// <summary>密码</summary>
public String Password { get; set; }
/// <summary>是否已登录</summary>
public Boolean Logined { get; set; }
/// <summary>最后一次登录成功后的消息</summary>
public LoginResponse Info { get; private set; }
#endregion
#region 方法
/// <summary>实例化</summary>
public AntClient()
{
Log = XTrace.Log;
//StatPeriod = 60;
ShowError = true;
//StatPeriod = 60;
ShowError = true;
#if DEBUG
EncoderLog = XTrace.Log;
StatPeriod = 10;
EncoderLog = XTrace.Log;
StatPeriod = 10;
#endif
}
/// <summary>实例化</summary>
/// <param name="uri"></param>
public AntClient(String uri) : this()
{
if (!uri.IsNullOrEmpty())
{
var ss = uri.Split(",", ";");
Servers = ss;
var u = new Uri(ss[0]);
var us = u.UserInfo.Split(":");
if (us.Length > 0) UserName = us[0];
if (us.Length > 1) Password = us[1];
}
}
#endregion
#region 登录
/// <summary>连接后自动登录</summary>
/// <param name="client">客户端</param>
/// <param name="force">强制登录</param>
protected override async Task<Object> OnLoginAsync(ISocketClient client, Boolean force)
{
if (Logined && !force) return null;
var asmx = AssemblyX.Entry;
var title = asmx?.Asm.GetCustomAttribute<AssemblyTitleAttribute>();
var dis = asmx?.Asm.GetCustomAttribute<DisplayNameAttribute>();
var des = asmx?.Asm.GetCustomAttribute<DescriptionAttribute>();
var dname = title?.Title ?? dis?.DisplayName ?? des?.Description;
var arg = new LoginModel
{
User = UserName,
Pass = Password.IsNullOrEmpty() ? null : Password.MD5(),
DisplayName = dname,
Machine = Environment.MachineName,
ProcessId = Process.GetCurrentProcess().Id,
Version = asmx.Version,
Compile = asmx.Compile,
};
var rs = await base.InvokeWithClientAsync<LoginResponse>(client, "Login", arg);
var set = AntSetting.Current;
if (set.Debug) XTrace.WriteLine("登录{0}成功！{1}", client, rs.ToJson());
// 保存下发密钥
if (!rs.Secret.IsNullOrEmpty())
{
set.Secret = rs.Secret;
set.Save();
}
Logined = true;
return Info = rs;
}
#endregion
#region 核心方法
/// <summary>获取指定名称的作业</summary>
/// <returns></returns>
public IJob[] GetJobs() => Invoke<JobModel[]>(nameof(GetJobs));
/// <summary>批量添加作业</summary>
/// <param name="jobs"></param>
/// <returns></returns>
public String[] AddJobs(IJob[] jobs) => Invoke<String[]>(nameof(AddJobs), new { jobs });
/// <summary>申请作业任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public ITask[] Acquire(String job, String topic, Int32 count) => Invoke<TaskModel[]>(nameof(Acquire), new AcquireModel
{
Job = job,
Topic = topic,
Count = count,
});
/// <summary>生产消息</summary>
/// <param name="model">模型</param>
/// <returns></returns>
public Int32 Produce(ProduceModel model) => Invoke<Int32>(nameof(Produce), model);
/// <summary>报告状态（进度、成功、错误）</summary>
/// <param name="task"></param>
/// <returns></returns>
public Boolean Report(ITaskResult task)
{
var retry = 3;
var lastex = new Exception();
while (retry-- > 0)
{
try
{
return Invoke<Boolean>(nameof(Report), task);
}
catch (Exception ex)
{
lastex = ex;
}
}
throw lastex;
}
/// <summary>获取当前应用的所有在线实例</summary>
/// <returns></returns>
public IPeer[] GetPeers() => Invoke<PeerModel[]>(nameof(GetPeers));
#endregion
}
/// <summary>实例化</summary>
/// <param name="uri"></param>
public AntClient(String uri) : this()
{
if (!uri.IsNullOrEmpty())
{
var ss = uri.Split(",", ";");
Servers = ss;
var u = new Uri(ss[0]);
var us = u.UserInfo.Split(":");
if (us.Length > 0) UserName = us[0];
if (us.Length > 1) Password = us[1];
}
}
#endregion
#region 登录
/// <summary>连接后自动登录</summary>
/// <param name="client">客户端</param>
/// <param name="force">强制登录</param>
protected override async Task<Object> OnLoginAsync(ISocketClient client, Boolean force)
{
if (Logined && !force) return null;
var asmx = AssemblyX.Entry;
var title = asmx?.Asm.GetCustomAttribute<AssemblyTitleAttribute>();
var dis = asmx?.Asm.GetCustomAttribute<DisplayNameAttribute>();
var des = asmx?.Asm.GetCustomAttribute<DescriptionAttribute>();
var dname = title?.Title ?? dis?.DisplayName ?? des?.Description;
var arg = new LoginModel
{
User = UserName,
Pass = Password.IsNullOrEmpty() ? null : Password.MD5(),
DisplayName = dname,
Machine = Environment.MachineName,
ProcessId = Process.GetCurrentProcess().Id,
Version = asmx.Version,
Compile = asmx.Compile,
};
var rs = await base.InvokeWithClientAsync<LoginResponse>(client, "Login", arg);
var set = AntSetting.Current;
if (set.Debug) XTrace.WriteLine("登录{0}成功！{1}", client, rs.ToJson());
// 保存下发密钥
if (!rs.Secret.IsNullOrEmpty())
{
set.Secret = rs.Secret;
set.Save();
}
Logined = true;
return Info = rs;
}
#endregion
#region 核心方法
/// <summary>获取指定名称的作业</summary>
/// <returns></returns>
public IJob[] GetJobs() => Invoke<JobModel[]>(nameof(GetJobs));
/// <summary>批量添加作业</summary>
/// <param name="jobs"></param>
/// <returns></returns>
public String[] AddJobs(IJob[] jobs) => Invoke<String[]>(nameof(AddJobs), new { jobs });
/// <summary>申请作业任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public ITask[] Acquire(String job, String topic, Int32 count) => Invoke<TaskModel[]>(nameof(Acquire), new AcquireModel
{
Job = job,
Topic = topic,
Count = count,
});
/// <summary>生产消息</summary>
/// <param name="model">模型</param>
/// <returns></returns>
public Int32 Produce(ProduceModel model) => Invoke<Int32>(nameof(Produce), model);
/// <summary>报告状态（进度、成功、错误）</summary>
/// <param name="task"></param>
/// <returns></returns>
public Boolean Report(ITaskResult task)
{
var retry = 3;
var lastex = new Exception();
while (retry-- > 0)
{
try
{
return Invoke<Boolean>(nameof(Report), task);
}
catch (Exception ex)
{
lastex = ex;
}
}
throw lastex;
}
/// <summary>获取当前应用的所有在线实例</summary>
/// <returns></returns>
public IPeer[] GetPeers() => Invoke<PeerModel[]>(nameof(GetPeers));
#endregion
}

424
AntJob/Providers/FileJobProvider.cs
View File

@ -1,230 +1,226 @@
using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Linq;
using System.ComponentModel;
using AntJob.Data;
using NewLife;
using NewLife.Log;
using NewLife.Reflection;
using NewLife.Xml;
namespace AntJob.Providers
namespace AntJob.Providers;
/// <summary>文件作业提供者</summary>
public class FileJobProvider : JobProvider
{
/// <summary>文件作业提供者</summary>
public class FileJobProvider : JobProvider
private JobFile _File;
/// <summary>销毁</summary>
/// <param name="disposing"></param>
protected override void Dispose(Boolean disposing)
{
private JobFile _File;
base.Dispose(disposing);
/// <summary>销毁</summary>
/// <param name="disposing"></param>
protected override void Dispose(Boolean disposing)
{
base.Dispose(disposing);
_File.TryDispose();
}
/// <summary>开始</summary>
public override void Start()
{
var jf = _File = JobFile.Current;
var list = new List<JobModel>();
if (jf.Jobs != null && jf.Jobs.Length > 0) list.AddRange(jf.Jobs);
// 扫描所有Worker并添加到作业文件
var flag = false;
foreach (var item in GetAll())
{
if (!list.Any(e => e.Name == item.Key))
{
// 新增作业项
var model = new JobModel();
// 获取默认设置
var job = item.Value.CreateInstance() as Handler;
var df = job?.Job;
if (df != null) model.Copy(df);
if (model.Start.Year <= 2000) model.Start = DateTime.Now.Date;
if (model.Step <= 0) model.Step = 30;
if (model.BatchSize <= 0) model.BatchSize = 10000;
if (model.MaxTask <= 0) model.MaxTask = Environment.ProcessorCount;
if (model.Name.IsNullOrEmpty())
{
model.Name = item.Key;
model.Enable = true;
}
list.Add(model);
flag = true;
}
}
if (flag)
{
if (jf.Jobs == null || jf.Jobs.Length == 0) jf.CreateTime = DateTime.Now;
jf.Jobs = list.ToArray();
}
jf.Save();
base.Start();
}
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public override IJob[] GetJobs()
{
var jf = _File = JobFile.Current;
var list = new List<IJob>();
if (jf.Jobs != null)
{
foreach (var item in jf.Jobs)
{
/*if (names.Contains(item.Name))*/
list.Add(item);
}
}
return list.ToArray();
}
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(IJob job, String topic, Int32 count)
{
var list = new List<ITask>();
if (!job.Enable) return list.ToArray();
// 当前时间减去偏移量，作为当前时间。数据抽取不许超过该时间
var now = DateTime.Now.AddSeconds(-job.Offset);
// 避免毫秒级带来误差，每毫秒有10000个滴答
var sec = now.Ticks / 1_000_0000;
now = new DateTime(sec * 1_000_0000);
var step = job.Step;
if (step <= 0) step = 30;
var start = job.Start;
for (var i = 0; i < count; i++)
{
// 开始时间和结束时间是否越界
if (start >= now) break;
var end = start.AddSeconds(step);
// 任务结束时间超过作业结束时间时，取后者
if (job.End.Year > 2000 && end > job.End) end = job.End;
// 时间片必须严格要求按照步进大小分片，除非有合适的End
if (job.Mode != JobModes.Alarm)
{
if (end > now) break;
}
// 时间区间判断
if (start >= end) break;
// 切分新任务
var set = new TaskModel
{
Start = start,
End = end,
//Step = job.Step,
//Offset = job.Offset,
BatchSize = job.BatchSize,
};
// 更新任务
job.Start = end;
start = end;
list.Add(set);
}
if (list.Count > 0)
{
_File.UpdateTime = DateTime.Now;
_File.SaveAsync();
}
return list.ToArray();
}
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public override void Finish(JobContext ctx)
{
var ex = ctx.Error?.GetTrue();
if (ex != null) XTrace.WriteException(ex);
if (ctx.Total > 0)
{
var set = ctx.Task;
var n = 0;
if (set.End > set.Start) n = (Int32)(set.End - set.Start).TotalSeconds;
var msg = $"{ctx.Handler.Name} 处理{ctx.Total:n0} 行，区间（{set.Start} + {n}, {set.End:HH:mm:ss}）";
if (ctx.Handler.Mode == JobModes.Alarm)
msg += $"，耗时{ctx.Cost:n0}ms";
else
msg += $"，速度{ctx.Speed:n0}tps，耗时{ctx.Cost:n0}ms";
XTrace.WriteLine(msg);
}
}
#region 静态扫描
private static IDictionary<String, Type> _workers;
/// <summary>扫描所有任务</summary>
public static IDictionary<String, Type> GetAll()
{
if (_workers != null) return _workers;
var dic = new Dictionary<String, Type>();
// 反射所有任务
foreach (var item in typeof(Handler).GetAllSubclasses())
{
if (item.IsAbstract) continue;
var name = item.GetDisplayName();
if (name.IsNullOrEmpty())
{
var wrk = item.CreateInstance() as Handler;
name = wrk?.Name;
}
if (name.IsNullOrEmpty()) name = item.FullName;
if (dic.TryGetValue(name, out var type))
XTrace.WriteLine("处理器[{0}]重复出现，Type1={1}，Type2={2}", name, type.FullName, item.FullName);
else
dic[name] = item;
}
return _workers = dic;
}
#endregion
_File.TryDispose();
}
/// <summary>作业配置</summary>
[Description("作业配置")]
[XmlConfigFile(@"Config\Job.config", 15000)]
public class JobFile : XmlConfig<JobFile>
/// <summary>开始</summary>
public override void Start()
{
/// <summary>创建时间</summary>
[Description("创建时间")]
public DateTime CreateTime { get; set; }
var jf = _File = JobFile.Current;
/// <summary>更新时间</summary>
[Description("更新时间")]
public DateTime UpdateTime { get; set; }
var list = new List<JobModel>();
if (jf.Jobs != null && jf.Jobs.Length > 0) list.AddRange(jf.Jobs);
/// <summary>作业集合</summary>
[Description("作业集合")]
public JobModel[] Jobs { get; set; }
// 扫描所有Worker并添加到作业文件
var flag = false;
foreach (var item in GetAll())
{
if (!list.Any(e => e.Name == item.Key))
{
// 新增作业项
var model = new JobModel();
// 获取默认设置
var job = item.Value.CreateInstance() as Handler;
var df = job?.Job;
if (df != null) model.Copy(df);
if (model.Start.Year <= 2000) model.Start = DateTime.Now.Date;
if (model.Step <= 0) model.Step = 30;
if (model.BatchSize <= 0) model.BatchSize = 10000;
if (model.MaxTask <= 0) model.MaxTask = Environment.ProcessorCount;
if (model.Name.IsNullOrEmpty())
{
model.Name = item.Key;
model.Enable = true;
}
list.Add(model);
flag = true;
}
}
if (flag)
{
if (jf.Jobs == null || jf.Jobs.Length == 0) jf.CreateTime = DateTime.Now;
jf.Jobs = list.ToArray();
}
jf.Save();
base.Start();
}
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public override IJob[] GetJobs()
{
var jf = _File = JobFile.Current;
var list = new List<IJob>();
if (jf.Jobs != null)
{
foreach (var item in jf.Jobs)
{
/*if (names.Contains(item.Name))*/
list.Add(item);
}
}
return list.ToArray();
}
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(IJob job, String topic, Int32 count)
{
var list = new List<ITask>();
if (!job.Enable) return list.ToArray();
// 当前时间减去偏移量，作为当前时间。数据抽取不许超过该时间
var now = DateTime.Now.AddSeconds(-job.Offset);
// 避免毫秒级带来误差，每毫秒有10000个滴答
var sec = now.Ticks / 1_000_0000;
now = new DateTime(sec * 1_000_0000);
var step = job.Step;
if (step <= 0) step = 30;
var start = job.Start;
for (var i = 0; i < count; i++)
{
// 开始时间和结束时间是否越界
if (start >= now) break;
var end = start.AddSeconds(step);
// 任务结束时间超过作业结束时间时，取后者
if (job.End.Year > 2000 && end > job.End) end = job.End;
// 时间片必须严格要求按照步进大小分片，除非有合适的End
if (job.Mode != JobModes.Alarm)
{
if (end > now) break;
}
// 时间区间判断
if (start >= end) break;
// 切分新任务
var set = new TaskModel
{
Start = start,
End = end,
//Step = job.Step,
//Offset = job.Offset,
BatchSize = job.BatchSize,
};
// 更新任务
job.Start = end;
start = end;
list.Add(set);
}
if (list.Count > 0)
{
_File.UpdateTime = DateTime.Now;
_File.SaveAsync();
}
return list.ToArray();
}
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public override void Finish(JobContext ctx)
{
var ex = ctx.Error?.GetTrue();
if (ex != null) XTrace.WriteException(ex);
if (ctx.Total > 0)
{
var set = ctx.Task;
var n = 0;
if (set.End > set.Start) n = (Int32)(set.End - set.Start).TotalSeconds;
var msg = $"{ctx.Handler.Name} 处理{ctx.Total:n0} 行，区间（{set.Start} + {n}, {set.End:HH:mm:ss}）";
if (ctx.Handler.Mode == JobModes.Alarm)
msg += $"，耗时{ctx.Cost:n0}ms";
else
msg += $"，速度{ctx.Speed:n0}tps，耗时{ctx.Cost:n0}ms";
XTrace.WriteLine(msg);
}
}
#region 静态扫描
private static IDictionary<String, Type> _workers;
/// <summary>扫描所有任务</summary>
public static IDictionary<String, Type> GetAll()
{
if (_workers != null) return _workers;
var dic = new Dictionary<String, Type>();
// 反射所有任务
foreach (var item in typeof(Handler).GetAllSubclasses())
{
if (item.IsAbstract) continue;
var name = item.GetDisplayName();
if (name.IsNullOrEmpty())
{
var wrk = item.CreateInstance() as Handler;
name = wrk?.Name;
}
if (name.IsNullOrEmpty()) name = item.FullName;
if (dic.TryGetValue(name, out var type))
XTrace.WriteLine("处理器[{0}]重复出现，Type1={1}，Type2={2}", name, type.FullName, item.FullName);
else
dic[name] = item;
}
return _workers = dic;
}
#endregion
}
/// <summary>作业配置</summary>
[Description("作业配置")]
[XmlConfigFile(@"Config\Job.config", 15000)]
public class JobFile : XmlConfig<JobFile>
{
/// <summary>创建时间</summary>
[Description("创建时间")]
public DateTime CreateTime { get; set; }
/// <summary>更新时间</summary>
[Description("更新时间")]
public DateTime UpdateTime { get; set; }
/// <summary>作业集合</summary>
[Description("作业集合")]
public JobModel[] Jobs { get; set; }
}

133
AntJob/Providers/IJobProvider.cs
View File

@ -1,87 +1,84 @@
using System;
using System.Collections.Generic;
using AntJob.Data;
using AntJob.Data;
using NewLife;
namespace AntJob.Providers
namespace AntJob.Providers;
/// <summary>作业提供者接口</summary>
public interface IJobProvider
{
/// <summary>作业提供者接口</summary>
public interface IJobProvider
{
/// <summary>调度器</summary>
Scheduler Schedule { get; set; }
/// <summary>调度器</summary>
Scheduler Schedule { get; set; }
/// <summary>开始工作</summary>
void Start();
/// <summary>开始工作</summary>
void Start();
/// <summary>停止工作</summary>
void Stop();
/// <summary>停止工作</summary>
void Stop();
/// <summary>获取所有作业。调度器定期获取以更新作业参数</summary>
/// <returns></returns>
IJob[] GetJobs();
/// <summary>获取所有作业。调度器定期获取以更新作业参数</summary>
/// <returns></returns>
IJob[] GetJobs();
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
ITask[] Acquire(IJob job, String topic, Int32 count);
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
ITask[] Acquire(IJob job, String topic, Int32 count);
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
Int32 Produce(String job, String topic, String[] messages, MessageOption option);
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
Int32 Produce(String job, String topic, String[] messages, MessageOption option);
/// <summary>报告进度</summary>
/// <param name="ctx">上下文</param>
void Report(JobContext ctx);
/// <summary>报告进度</summary>
/// <param name="ctx">上下文</param>
void Report(JobContext ctx);
/// <summary>完成任务</summary>
/// <param name="ctx">上下文</param>
void Finish(JobContext ctx);
}
/// <summary>完成任务</summary>
/// <param name="ctx">上下文</param>
void Finish(JobContext ctx);
}
/// <summary>任务提供者基类</summary>
public abstract class JobProvider : DisposeBase, IJobProvider
{
/// <summary>调度器</summary>
public Scheduler Schedule { get; set; }
/// <summary>任务提供者基类</summary>
public abstract class JobProvider : DisposeBase, IJobProvider
{
/// <summary>调度器</summary>
public Scheduler Schedule { get; set; }
/// <summary>开始工作</summary>
public virtual void Start() { }
/// <summary>开始工作</summary>
public virtual void Start() { }
/// <summary>停止工作</summary>
public virtual void Stop() { }
/// <summary>停止工作</summary>
public virtual void Stop() { }
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public abstract IJob[] GetJobs();
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public abstract IJob[] GetJobs();
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public abstract ITask[] Acquire(IJob job, String topic, Int32 count);
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public abstract ITask[] Acquire(IJob job, String topic, Int32 count);
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public virtual Int32 Produce(String job, String topic, String[] messages, MessageOption option = null) => 0;
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public virtual Int32 Produce(String job, String topic, String[] messages, MessageOption option = null) => 0;
/// <summary>报告进度，每个任务多次调用</summary>
/// <param name="ctx">上下文</param>
public virtual void Report(JobContext ctx) { }
/// <summary>报告进度，每个任务多次调用</summary>
/// <param name="ctx">上下文</param>
public virtual void Report(JobContext ctx) { }
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public virtual void Finish(JobContext ctx) { }
}
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public virtual void Finish(JobContext ctx) { }
}

17
AntJob/Providers/MessageOption.cs
View File

@ -1,14 +1,11 @@
using System;
namespace AntJob.Providers;
namespace AntJob.Providers
/// <summary>消息选项</summary>
public class MessageOption
{
/// <summary>消息选项</summary>
public class MessageOption
{
/// <summary>延迟执行间隔（实际执行时间=延迟+生产时间），单位秒</summary>
public Int32 DelayTime { get; set; }
/// <summary>延迟执行间隔（实际执行时间=延迟+生产时间），单位秒</summary>
public Int32 DelayTime { get; set; }
/// <summary>消息去重。避免单个消息被重复生产</summary>
public Boolean Unique { get; set; }
}
/// <summary>消息去重。避免单个消息被重复生产</summary>
public Boolean Unique { get; set; }
}

493
AntJob/Providers/NetworkJobProvider.cs
View File

@ -1,257 +1,258 @@
using System;
using System.Collections.Generic;
using System.Linq;
using AntJob.Data;
using AntJob.Data;
using AntJob.Handlers;
using AntJob.Models;
using NewLife;
using NewLife.Log;
using NewLife.Threading;
namespace AntJob.Providers
namespace AntJob.Providers;
/// <summary>网络任务提供者</summary>
public class NetworkJobProvider : JobProvider
{
/// <summary>网络任务提供者</summary>
public class NetworkJobProvider : JobProvider
#region 属性
/// <summary>调试，打开编码日志</summary>
public Boolean Debug { get; set; }
/// <summary>调度中心地址</summary>
public String Server { get; set; }
/// <summary>应用编号</summary>
public String AppID { get; set; }
/// <summary>应用密钥</summary>
public String Secret { get; set; }
/// <summary>客户端</summary>
public AntClient Ant { get; set; }
/// <summary>邻居伙伴。用于应用判断自身有多少个实例在运行</summary>
public IPeer[] Peers { get; private set; }
/// <summary>性能跟踪器</summary>
public ITracer Tracer { get; set; }
#endregion
#region 构造
/// <summary>销毁</summary>
/// <param name="disposing"></param>
protected override void Dispose(Boolean disposing)
{
#region 属性
/// <summary>调试，打开编码日志</summary>
public Boolean Debug { get; set; }
base.Dispose(disposing);
/// <summary>调度中心地址</summary>
public String Server { get; set; }
/// <summary>应用编号</summary>
public String AppID { get; set; }
/// <summary>应用密钥</summary>
public String Secret { get; set; }
/// <summary>客户端</summary>
public AntClient Ant { get; set; }
/// <summary>邻居伙伴</summary>
public IPeer[] Peers { get; private set; }
#endregion
#region 构造
/// <summary>销毁</summary>
/// <param name="disposing"></param>
protected override void Dispose(Boolean disposing)
{
base.Dispose(disposing);
_timer.TryDispose();
_timer = null;
}
#endregion
#region 启动停止
/// <summary>开始</summary>
public override void Start()
{
var svr = Server;
// 使用配置中心账号
var ant = new AntClient(svr)
{
UserName = AppID,
Password = Secret
};
if (Debug) ant.EncoderLog = XTrace.Log;
ant.Open();
// 断开前一个连接
Ant.TryDispose();
Ant = ant;
var bs = Schedule?.Handlers;
//var jobs = GetJobs(ws.Select(e => e.Name).ToArray());
var list = new List<IJob>();
foreach (var handler in bs)
{
var job = handler.Job ?? new JobModel();
job.Name = handler.Name;
job.ClassName = handler.GetType().FullName;
job.Mode = handler.Mode;
// 描述
if (job is JobModel job2)
{
var dis = handler.GetType().GetDisplayName();
if (!dis.IsNullOrEmpty()) job2.DisplayName = dis;
var des = handler.GetType().GetDescription();
if (!des.IsNullOrEmpty()) job2.Description = des;
if (handler is MessageHandler mhandler) job2.Topic = mhandler.Topic;
}
list.Add(job);
}
if (list.Count > 0) Ant.AddJobs(list.ToArray());
// 定时更新邻居
_timer = new TimerX(DoCheckPeer, null, 1_000, 30_000) { Async = true };
}
/// <summary>停止</summary>
public override void Stop()
{
// 断开前一个连接
Ant.TryDispose();
Ant = null;
}
#endregion
#region 作业消息控制
private IJob[] _jobs;
private DateTime _NextGetJobs;
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public override IJob[] GetJobs()
{
// 周期性获取，避免请求过快
var now = TimerX.Now;
if (_jobs == null || _NextGetJobs <= now)
{
_NextGetJobs = now.AddSeconds(5);
_jobs = Ant.GetJobs();
}
return _jobs;
}
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(IJob job, String topic, Int32 count) => Ant.Acquire(job.Name, topic, count);
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public override Int32 Produce(String job, String topic, String[] messages, MessageOption option = null)
{
if (topic.IsNullOrEmpty() || messages == null || messages.Length < 1) return 0;
var model = new ProduceModel
{
Job = job,
Topic = topic,
Messages = messages,
};
if (option != null)
{
model.DelayTime = option.DelayTime;
model.Unique = option.Unique;
}
return Ant.Produce(model);
}
#endregion
#region 报告状态
//private static readonly String _MachineName = Environment.MachineName;
//private static readonly Int32 _ProcessID = Process.GetCurrentProcess().Id;
/// <summary>报告进度，每个任务多次调用</summary>
/// <param name="ctx">上下文</param>
public override void Report(JobContext ctx)
{
// 不用上报抽取中
if (ctx.Status == JobStatus.抽取中) return;
if (ctx?.Result is not TaskResult task) return;
// 区分抽取和处理
task.Status = ctx.Status;
task.Speed = ctx.Speed;
task.Total = ctx.Total;
task.Success = ctx.Success;
Report(ctx.Handler.Job, task);
}
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public override void Finish(JobContext ctx)
{
if (ctx?.Result is not TaskResult task) return;
task.Speed = ctx.Speed;
task.Total = ctx.Total;
task.Success = ctx.Success;
task.Times++;
// 区分正常完成还是错误终止
if (ctx.Error != null)
{
task.Error++;
task.Status = JobStatus.错误;
var ex = ctx.Error?.GetTrue();
if (ex != null)
{
var msg = ctx.Error.GetMessage();
if (msg.Contains("Exception:")) msg = msg.Substring("Exception:").Trim();
task.Message = msg;
}
}
else
{
task.Status = JobStatus.完成;
task.Cost = (Int32)Math.Round(ctx.Cost / 1000);
}
if (task.Message.IsNullOrEmpty()) task.Message = ctx.Remark;
task.Key = ctx.Key;
Report(ctx.Handler.Job, task);
}
private void Report(IJob job, ITaskResult task)
{
try
{
Ant.Report(task);
}
catch (Exception ex)
{
XTrace.WriteLine("[{0}]的[{1}]状态报告失败！{2}", job, task.Status, ex.GetTrue().Message);
}
}
#endregion
#region 邻居
private TimerX _timer;
private void DoCheckPeer(Object state)
{
var ps = Ant?.GetPeers();
if (ps == null || ps.Length == 0) return;
var old = (Peers ?? new IPeer[0]).ToList();
foreach (var item in ps)
{
var pr = old.FirstOrDefault(e => e.Instance == item.Instance);
if (pr == null)
XTrace.WriteLine("[{0}]上线！{1}", item.Instance, item.Machine);
else
old.Remove(pr);
}
foreach (var item in old)
{
XTrace.WriteLine("[{0}]下线！{1}", item.Instance, item.Machine);
}
Peers = ps;
}
#endregion
_timer.TryDispose();
_timer = null;
}
#endregion
#region 启动停止
/// <summary>开始</summary>
public override void Start()
{
var svr = Server;
// 使用配置中心账号
var ant = new AntClient(svr)
{
UserName = AppID,
Password = Secret,
Tracer = Tracer,
};
if (Debug) ant.EncoderLog = XTrace.Log;
ant.Open();
// 断开前一个连接
Ant.TryDispose();
Ant = ant;
var bs = Schedule?.Handlers;
// 遍历所有处理器，添加作业到调度中心
//var jobs = GetJobs(ws.Select(e => e.Name).ToArray());
var list = new List<IJob>();
foreach (var handler in bs)
{
var job = handler.Job ?? new JobModel();
job.Name = handler.Name;
job.ClassName = handler.GetType().FullName;
job.Mode = handler.Mode;
// 描述
if (job is JobModel job2)
{
var dis = handler.GetType().GetDisplayName();
if (!dis.IsNullOrEmpty()) job2.DisplayName = dis;
var des = handler.GetType().GetDescription();
if (!des.IsNullOrEmpty()) job2.Description = des;
if (handler is MessageHandler mhandler) job2.Topic = mhandler.Topic;
}
list.Add(job);
}
if (list.Count > 0) Ant.AddJobs(list.ToArray());
// 定时更新邻居
_timer = new TimerX(DoCheckPeer, null, 1_000, 30_000) { Async = true };
}
/// <summary>停止</summary>
public override void Stop()
{
// 断开前一个连接
Ant.TryDispose();
Ant = null;
}
#endregion
#region 作业消息控制
private IJob[] _jobs;
private DateTime _NextGetJobs;
/// <summary>获取所有作业名称</summary>
/// <returns></returns>
public override IJob[] GetJobs()
{
// 周期性获取，避免请求过快
var now = TimerX.Now;
if (_jobs == null || _NextGetJobs <= now)
{
_NextGetJobs = now.AddSeconds(5);
_jobs = Ant.GetJobs();
}
return _jobs;
}
/// <summary>申请任务</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="count">要申请的任务个数</param>
/// <returns></returns>
public override ITask[] Acquire(IJob job, String topic, Int32 count) => Ant.Acquire(job.Name, topic, count);
/// <summary>生产消息</summary>
/// <param name="job">作业</param>
/// <param name="topic">主题</param>
/// <param name="messages">消息集合</param>
/// <param name="option">消息选项</param>
/// <returns></returns>
public override Int32 Produce(String job, String topic, String[] messages, MessageOption option = null)
{
if (topic.IsNullOrEmpty() || messages == null || messages.Length < 1) return 0;
var model = new ProduceModel
{
Job = job,
Topic = topic,
Messages = messages,
};
if (option != null)
{
model.DelayTime = option.DelayTime;
model.Unique = option.Unique;
}
return Ant.Produce(model);
}
#endregion
#region 报告状态
//private static readonly String _MachineName = Environment.MachineName;
//private static readonly Int32 _ProcessID = Process.GetCurrentProcess().Id;
/// <summary>报告进度，每个任务多次调用</summary>
/// <param name="ctx">上下文</param>
public override void Report(JobContext ctx)
{
// 不用上报抽取中
if (ctx.Status == JobStatus.抽取中) return;
if (ctx?.Result is not TaskResult task) return;
// 区分抽取和处理
task.Status = ctx.Status;
task.Speed = ctx.Speed;
task.Total = ctx.Total;
task.Success = ctx.Success;
Report(ctx.Handler.Job, task);
}
/// <summary>完成任务，每个任务只调用一次</summary>
/// <param name="ctx">上下文</param>
public override void Finish(JobContext ctx)
{
if (ctx?.Result is not TaskResult task) return;
task.Speed = ctx.Speed;
task.Total = ctx.Total;
task.Success = ctx.Success;
task.Times++;
// 区分正常完成还是错误终止
if (ctx.Error != null)
{
task.Error++;
task.Status = JobStatus.错误;
var ex = ctx.Error?.GetTrue();
if (ex != null)
{
var msg = ctx.Error.GetMessage();
if (msg.Contains("Exception:")) msg = msg.Substring("Exception:").Trim();
task.Message = msg;
}
}
else
{
task.Status = JobStatus.完成;
task.Cost = (Int32)Math.Round(ctx.Cost / 1000);
}
if (task.Message.IsNullOrEmpty()) task.Message = ctx.Remark;
task.Key = ctx.Key;
Report(ctx.Handler.Job, task);
}
private void Report(IJob job, ITaskResult task)
{
try
{
Ant.Report(task);
}
catch (Exception ex)
{
XTrace.WriteLine("[{0}]的[{1}]状态报告失败！{2}", job, task.Status, ex.GetTrue().Message);
}
}
#endregion
#region 邻居
private TimerX _timer;
private void DoCheckPeer(Object state)
{
var ps = Ant?.GetPeers();
if (ps == null || ps.Length == 0) return;
var old = (Peers ?? new IPeer[0]).ToList();
foreach (var item in ps)
{
var pr = old.FirstOrDefault(e => e.Instance == item.Instance);
if (pr == null)
XTrace.WriteLine("[{0}]上线！{1}", item.Instance, item.Machine);
else
old.Remove(pr);
}
foreach (var item in old)
{
XTrace.WriteLine("[{0}]下线！{1}", item.Instance, item.Machine);
}
Peers = ps;
}
#endregion
}

34
AntJob/Scheduler.cs
View File

@ -63,20 +63,27 @@ public class Scheduler : DisposeBase
var hs = Handlers;
if (hs.Count == 0) throw new ArgumentNullException(nameof(Handlers), "没有可用处理器");
// 埋点
using var span = Tracer?.NewSpan("job:SchedulerStart");
// 启动作业提供者
var prv = Provider;
prv ??= Provider = new FileJobProvider();
prv.Schedule ??= this;
// 从注册中心获取包
if (prv is NetworkJobProvider network && network.Server.IsNullOrEmpty())
if (prv is NetworkJobProvider network)
{
var registry = ServiceProvider?.GetService<IRegistry>();
if (registry != null)
{
var svrs = registry.ResolveAddressAsync("AntServer").Result;
network.Tracer ??= Tracer;
if (svrs != null && svrs.Length > 0) network.Server = svrs.Join();
if (network.Server.IsNullOrEmpty())
{
// 从注册中心获取包
var registry = ServiceProvider?.GetService<IRegistry>();
if (registry != null)
{
var svrs = registry.ResolveAddressAsync("AntServer").Result;
if (svrs != null && svrs.Length > 0) network.Server = svrs.Join();
}
}
}
@ -112,6 +119,8 @@ public class Scheduler : DisposeBase
/// <summary>停止</summary>
public void Stop()
{
using var span = Tracer?.NewSpan("job:SchedulerStop");
_timer.TryDispose();
_timer = null;
@ -119,7 +128,7 @@ public class Scheduler : DisposeBase
foreach (var handler in Handlers)
{
handler.Stop();
handler.Stop("SchedulerStop");
}
}
@ -147,7 +156,7 @@ public class Scheduler : DisposeBase
// 找不到或者已停用
if (job == null || !job.Enable)
{
if (handler.Active) handler.Stop();
if (handler.Active) handler.Stop("ConfigChanged");
continue;
}
@ -201,12 +210,13 @@ public class Scheduler : DisposeBase
var handler = handlers.FirstOrDefault(e => e.Name == job.Name);
if (handler == null && job.Enable && !job.ClassName.IsNullOrEmpty())
{
using var span = Tracer?.NewSpan($"job:NewHandler", job);
XTrace.WriteLine("发现未知作业[{0}]@[{1}]", job.Name, job.ClassName);
try
{
// 实例化一个处理器
var type = Type.GetType(job.ClassName);
if (type == null) type = handlers.Where(e => e.GetType().FullName == job.ClassName)?.FirstOrDefault()?.GetType();
var type = Type.GetType(job.ClassName) ?? (handlers.Where(e => e.GetType().FullName == job.ClassName)?.FirstOrDefault()?.GetType());
if (type != null)
{
handler = type.CreateInstance() as Handler;
@ -221,6 +231,7 @@ public class Scheduler : DisposeBase
if (handler is MessageHandler messageHandler && !job.Topic.IsNullOrEmpty()) messageHandler.Topic = job.Topic;
handler.Log = XTrace.Log;
handler.Tracer = Tracer;
handler.Start();
handlers.Add(handler);
@ -229,6 +240,7 @@ public class Scheduler : DisposeBase
}
catch (Exception ex)
{
span?.SetError(ex, null);
XTrace.WriteException(ex);
}
}

2
AntTest/AntTest.csproj
View File

@ -4,7 +4,7 @@
<TargetFramework>net6.0</TargetFramework>
<OutputPath>..\Bin\UnitTest</OutputPath>
<AppendTargetFrameworkToOutputPath>false</AppendTargetFrameworkToOutputPath>
<LangVersion>latest</LangVersion>
<IsPackable>false</IsPackable>
</PropertyGroup>

70
AntTest/SqlHandlerTests.cs
View File

@ -2,51 +2,53 @@
using System.Linq;
using AntJob;
using AntJob.Data;
using AntJob.Extensions;
using NewLife.Reflection;
using XCode.DataAccessLayer;
using XCode.Membership;
using Xunit;
namespace AntTest
namespace AntTest;
public class SqlHandlerTests
{
public class SqlHandlerTests
[Fact]
public void ParseAllRoles()
{
[Fact]
public void ParseAllRoles()
var tt = """
/*use membership*/
select * from role
/*use membership_bak*/
delete from role;
/*use membership_bak*/
insert role;
""";
var count = Role.Meta.Count;
var table = Role.Meta.Session.Dal.Tables.FirstOrDefault(e => e.Name == "Role");
DAL.Create("membership").SetTables(table);
DAL.Create("membership_bak").SetTables(table);
var handler = new SqlHandler();
var task = new TaskModel
{
var tt = @"/*use membership*/
select * from role
Data = tt
};
/*use membership_bak*/
delete from role;
//handler.Process(task);
/*use membership_bak*/
insert role;
";
var count = Role.Meta.Count;
var table = Role.Meta.Session.Dal.Tables.FirstOrDefault(e => e.Name == "Role");
DAL.Create("membership").SetTables(table);
DAL.Create("membership_bak").SetTables(table);
var ctx = new JobContext
{
Task = task,
};
var handler = new SqlHandler();
var task = new TaskModel
{
Data = tt
};
var method = handler.GetType().GetMethodEx("OnProcess", typeof(JobContext));
method.Invoke(handler, new Object[] { ctx });
//handler.Process(task);
var ctx = new JobContext
{
Task = task,
};
var method = handler.GetType().GetMethodEx("OnProcess", typeof(JobContext));
method.Invoke(handler, new Object[] { ctx });
Assert.Equal(4, ctx.Total);
//Assert.Equal(4, ctx.Success);
Assert.True(ctx.Success > 0);
}
Assert.Equal(4, ctx.Total);
//Assert.Equal(4, ctx.Success);
Assert.True(ctx.Success > 0);
}
}

1
Samples/HisAgent/BuildWill.cs
View File

@ -1,6 +1,7 @@
using System;
using System.ComponentModel;
using AntJob;
using AntJob.Extensions;
using HisData;
using NewLife.Security;
using XCode;