首页 > 其他分享 >开源:Taurus.DTS 微服务分布式任务框架,支持即时任务、延时任务、Cron表达式定时任务和广播任务。

开源:Taurus.DTS 微服务分布式任务框架,支持即时任务、延时任务、Cron表达式定时任务和广播任务。

时间:2024-01-18 14:34:18浏览次数:35  
标签:0.1 Taurus Server Cron 任务 Client DTS

前言:

在发布完:开源:Taurus.DTC 微服务分布式事务框架,支持 .Net 和 .Net Core 双系列版本,之后想想,好像除了事务外,感觉里面多了一个任务发布订阅的基础功能。

本想既然都有了基础发布订阅功能了,那要不要顺带加上延时发布功能呢?加上了会不会让事务组件不纯了?

经过一翻深思,是在其上补上功能,还是,重新写一个组件,起初起名是个难题,因为 DTC 也可以是 Distributed Task Cxxxxx, 组件重名了?

经过一翻英文大作战,找到了:Distributed Task Scheduler,简写可以是DTS了,才开始重启一个组件。

于是就有了这个Taurus.DTS 任务组件,而且功能除了原有的即时任务发布订阅,和本来想加上的延时任务,后面又补上了基于Cron表达式的定时任务,和广播群发任务。

经过一翻大作战,有 Taurus.DTC 的基础代码作底层支持,花了大几天,解决了各种疑难杂症之后,终于出来了。

1、开源地址:

https://github.com/cyq1162/Taurus.DTS

2、Nuget 包引用ID:

由于 CYQ.Data Orm 组件本身支持10多种数据库,因此提供的包,只根据消息队列的需要分拆提供。

默认Taurus.DTS 支持同时使用 RabbitMQ 和 Kafka 两种消息队列。

如果一个项目中只用RabbitMQ,则引入 Tarurus.DTS.RabbitMQ,减少对 Kafka 的依赖包。

编绎的版本:支持太多,发布是个苦力活:

.Net 系列:由于引用依赖关系(RabbitMQ最低是 4.0,Kafka最低是 4.5);

.Net Core系列、支持 2.1 到 8.0 版本及以后。

Standard 标准库:支持2.1 及以后。

3、Taurus.DTS 微服务分布式任务框架基础说明:

基础说明:

1、框架分为: Client(客户端,即任务发起端)和 Server(服务端,即方法订阅方)。

2、框架支持:即时任务、延时任务、Cron表达式任务定时任务、广播任务,四种方式。

3、项目需要配置的参数:1、数据库(可选);2、MQ(必选)。

数据存储:

可选择数据库(MSSQL、MySql、Oracle、PostgreSql 等 CYQ.Data 所支持的10多种数据库之一)

MSSQL配置示例如下:

{
  "ConnectionStrings": {
    "DTS.Server.Conn": "server=.;database=MSLog;uid=sa;pwd=123456"
  }
}
 

消息队列:

目前消息队列支持 RabbitMQ 或者 Kafka(配置其中一方即可):

{
  "AppSettings": {
  "DTS.Server.Rabbit":"127.0.0.1;guest;guest;/",//ip;username;password;virtualpath;
  "DTS.Server.Kafka":"127.0.0.1:9092" 
  }
}

以上配置为Server端,客户端更改 Server 为 Client 即可。

4、Server 端 使用示例:

1、Nuget 搜索 Taurus.DTS 引入工程项目中。

2、如果是 ASP.Net Core 程序:Program 或 Startup 添加服务使用引入:

  services.AddTaurusDts(); // 服务添加。
  app.UseTaurusDts(TaskStartType.Server); //服务使用,启用服务端

3、appsettings.json 配置基本属性:

  {
  "ConnectionStrings": {
    "DTS.Server.Conn": "host=localhost;port=3306;database=cyqdata;uid=root;pwd=123456;Convert Zero Datetime=True;"
  },
  "AppSettings": {
    "DTS.Server.Rabbit": "127.0.0.1;guest;guest;/" //IP;UserName;Password;VirtualPaath
}

4、选择数据库对应的依赖组件,如MySql,可以:

Nuget 上可以搜索 MySql.Data 、或者 CYQ.Data.MySql (会自动引入MySql.Data)  都可, 引入项目即可。

5、代码编写,可以参考源码中提供的示例代码,如下为控制台示例代码:

using System;
using Taurus.Plugin.DistributedTask;
namespace Console_App_Server {

 internal class Program
 {
 
     static void Main(string[] args)
    {

        DTSConfig.Server.Rabbit = "127.0.0.1;guest;guest;/";
        //DTSConfig.Server.Kafka = "127.0.0.1:9092;";
        //DTSConfig.Server.Conn = DTSConfig.Client.Conn;

        DTSConfig.ProjectName = "ConsoleApp5";

        DTS.Server.Start();//start client and server

        Console.WriteLine("---------------------------------------");

        Console.ReadLine();
    }


}


/// <summary>
/// 服务端 server class need to public
/// </summary>
public class Server
{
    [DTSSubscribe("DoInstantTask")]
    public static bool A(DTSSubscribePara para)
    {
        para.CallBackContent = "show you a.";
        return true;
    }

    [DTSSubscribe("DoDelayTask")]
    private static bool B(DTSSubscribePara para)
    {
        para.CallBackContent = "show you b.";
        return true;
    }
    [DTSSubscribe("DoCronTask")]
    private static bool C(DTSSubscribePara para)
    {
        para.CallBackContent = "show you c.";
        return true;
    }
    /// <summary>
    /// 定时任务
    /// </summary>
    [DTSSubscribe("DoBroadastTask")]
    private static bool TimerTask(DTSSubscribePara para)
    {
        para.CallBackContent = "show you d.";
        return true;
    }
}
 
}

5、Client 端 使用示例:

1、Nuget 搜索 Taurus.DTS 引入工程项目中。

2、如果是ASP.Net Core 程序:Program 或 Startup 添加服务使用引入:

  services.AddTaurusDts(); // 服务添加
  app.UseTaurusDts(StartType.Client); //服务使用,启用服务端

3、appsettings.json 配置基本属性:

  {
  "ConnectionStrings": {
    "DTS.Client.Conn": "host=localhost;port=3306;database=cyqdata;uid=root;pwd=123456;Convert Zero Datetime=True;"
  },
  "AppSettings": {
    "DTS.Client.Rabbit": "127.0.0.1;guest;guest;/" //IP;UserName;Password;VirtualPaath
}

4、选择数据库对应的依赖组件,如MySql,可以:

Nuget 上可以搜索 MySql.Data 、或者 CYQ.Data.MySql (会自动引入MySql.Data)  都可, 引入项目即可。

5、代码编写,可以参考源码中提供的示例代码,如下为控制台示例代码:

using System;
using System.Threading;
using Taurus.Plugin.DistributedTask;
namespace Console_App_Client {

  internal class Program
  {
  
    static void Main(string[] args)
    {
    
        DTSConfig.Client.IsPrintTraceLog = false;
        //AppConfig.Redis.Servers = "127.0.0.1:6379";

        DTSConfig.Client.Rabbit = "127.0.0.1;guest;guest;/";
        //DTSConfig.Client.Kafka = "127.0.0.1:9092;";
        DTSConfig.Client.Conn = "server=.;database=mslog;uid=sa;pwd=123456";

        DTSConfig.ProjectName = "ConsoleApp5";

        DTS.Client.Start();//start client and server
        
        Console.WriteLine("---------------------------------------");
        Console.WriteLine("1-InstantTask、2-DelayTask(1Minutes)、3-CronTask、4-DeleteCronTask、5-BroadastTask");
        Console.WriteLine("Input :1、2、3、4、5,Press Enter.");
        while (true)
        {
            string line = Console.ReadLine();
            try
            {
                Client.Run(int.Parse(line));
            }
            catch(Exception err)
            {
                Console.WriteLine(err.Message);
            }
            
        }

    }
}

/// <summary>
/// 客户端 client class need to public if has callback method.
/// </summary>
public class Client
{
    public static void Run(int i)
    {

        if (i == 2)
        {
            //发布一个延时1分钟的任务
            DTS.Client.Delay.PublishAsync(1, "i publish a delay task.", "DoDelayTask", "DelayCallBack");
            Console.WriteLine("Wait for 1 minute...");
        }
        else if (i == 3)
        {
            //发布一个秒在30时的循环任务。
            DTS.Client.Cron.PublishAsync("10,30,50 * * * * ?", "i publish a timer task with cron express.", "DoCronTask", "CronCallBack");
            Console.WriteLine("Wait for execute task when second is 10,30,50...");
        }
        else if (i == 4)
        {
            //发布一个秒在30时的循环任务。
            DTS.Client.Cron.DeleteAsync("DoCronTask", null, "CronCallBack");
        }
        else if (i == 5)
        {
            //发布一个广播任务
            DTS.Client.Broadast.PublishAsync("i publish a task for all server.", "DoBroadastTask", "BroadastCallBack");
        }
        else
        {
            for (int k = 0; k < 1; k++)
            {
                //发布一个即时任务
                DTS.Client.Instant.PublishAsync("i publish a task instantly.", "DoInstantTask", "InstantCallBack");
                Console.WriteLine(k);
            }
            
        }
    }

    [DTSCallBack("InstantCallBack")]
    [DTSCallBack("DelayCallBack")]
    [DTSCallBack("CronCallBack")]
    [DTSCallBack("BroadastCallBack")]
    private static void OnCallBack(DTSCallBackPara para)
    {
        Console.WriteLine("Client callback : " + para.TaskType + " - " + para.CallBackKey + " - " + para.CallBackContent);
    }
}
 
}

6、Demo 运行示例:

demo 地址:https://github.com/cyq1162/Taurus.DTS/tree/master/demo

启动运行截图:

 输入1,发布即时任务:

7、其它:CYQ.Data 支持的数据库链接语句示例

###--------------------------------------------------------###

   Txt::  Txt Path=E:\
   Xml::  Xml Path=E:\
Access::  Provider=Microsoft.Jet.OLEDB.4.0; Data Source=E:\cyqdata.mdb
Sqlite::  Data Source=E:\cyqdata.db;failifmissing=false;
 MySql::  host=localhost;port=3306;database=cyqdata;uid=root;pwd=123456;Convert Zero Datetime=True;
 Mssql::  server=.;database=cyqdata;uid=sa;pwd=123456;provider=mssql; 
Sybase::  data source=127.0.0.1;port=5000;database=cyqdata;uid=sa;pwd=123456;provider=sybase; 
Postgre:  server=localhost;uid=sa;pwd=123456;database=cyqdata;provider=pg; 
    DB2:  Database=SAMPLE;User ID=administrator;Server=127.0.0.1;password=1234560;provider=db2; 
FireBird  user id=SYSDBA;password=123456;database=d:\\test.dbf;server type=Default;data source=127.0.0.1;port number=3050;provider=firebird;
Dameng::  user id=SYSDBA;password=123456789;data source=127.0.0.1;schema=test;provider=dameng;
KingBaseES server=127.0.0.1;User Id=system;Password=123456;Database=test;Port=54321;schema=public;provider=kingbasees;
Oracle ODP.NET::
Data Source=(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)(HOST=127.0.0.1)(PORT = 1521)))(CONNECT_DATA =(SID = orcl)));User ID=sa;password=123456

由于各种数据库链接语句基本一致,除了特定写法外,可以通过链接补充:provider=mssql、provider=mysql、provider=db2、provider=postgre等来区分。
###--------------------------------------------------------###

8、总结:

由于 Taurus.DTS 分布式任务发布组件的独立发布,原有发布的 Taurus.DTC 分布式事务组件,下一版本会移除掉其基础的任务发布订阅功能,保留事务的纯洁属性。

今天发布的 Taurus.DTS 组件,为 .Net 和  .Net Core 微服务系列又又贡献了一个新的组件。

后续会发布分布式锁的教程,这个已经在 CYQ.Data 里实现了,并且在也在 Taurus.DTC 和 Taurus.DTS 中使用到了。

标签:0.1,Taurus,Server,Cron,任务,Client,DTS
From: https://www.cnblogs.com/cyq1162/p/17971549

相关文章

  • 奇迹Mu中冒险家们的新任务——挑战副本
    奇迹Mu作为一款经典的多人在线角色扮演游戏,在其丰富多彩的游戏世界中,紧跟潮流的游戏开发者们也不断推出更新和新玩法,以吸引更多的玩家加入其中。不久前,奇迹Mu更新了一项全新玩法——挑战副本。这项新任务不仅为冒险家们带来了更多的挑战和乐趣,更为游戏的发展注入了新的活力。以下......
  • Linux 中 Crontab 执行时的环境变量问题(allure命令不执行)
    前几天做了UI自动化脚本部署linux服务器,但是放下脚本的allure命令不执行(生成allure报告和启动allure服务的命令不执行),然后就各种找问题,一开始怀疑是allure的环境变量问题,但是发现使用allure--version可以正常显示allure的版本没问题后边尝试将环境变量加入linuxcronta......
  • 计划任务 显示消息 已弃用 计划任务弹出提示
    简介:计划任务的显示消息已弃用了。但是我这里需要一个每周一早上8点重启的任务,太多懒人不关电脑了,更新不执行,稳定下降,很多需要重启才能完成的功能均不正常,只好计划任务每周重启一下吧。但是各个终端用户又有着不同的使用习惯:长期开着excel不关,不保存……组策略定时重启吧。解......
  • Celery 任务路由的使用,在多任务时,实现分组管理任务
    Celery任务路由的使用,在多任务时,实现分组管理任务Celery任务路由的使用,本文参考了国外大佬的文章,并做了修改与补充,原文见这里,BjoernStielCelery官方文档:RoutingTasks任务路由的主要作用:把任务分类,让他们在不同队列中运行,互不干扰,同时方便管理本文介绍3种任务路由添加......
  • crontab 环境变量的问题
    1、大致功能就是找出未在fstab中挂载的磁盘,并发送到云平台告警。#!/bin/bashsource/etc/profileremote_address="http://1.1.1.1:6666"user=xxxxpass=xxxxexcept_block=("/dev/sr0""/dev/cdrom""/dev/dvd")blkid_output=$(blkid-sUUID)all_blo......
  • sqlserver查询最近失败的任务
    selectjob_id,step_name,message,cast((cast(LEFT(run_date,4)ASVARCHAR)+'-'+SUBSTRING(cast(run_dateASVARCHAR),5,2)+'-'+cast(RIGHT(run_date,2)ASVARCHAR))+'......
  • 时光机启动:Spring中如何巧妙实现定时任务?
    嗨,亲爱的小伙伴们!小米在这里又来和大家分享一些技术干货啦!今天我们要探讨的话题是关于Spring框架中如何实现定时任务。对于我们这些热爱技术的小伙伴来说,定时任务可是一个非常有趣而且实用的话题哦!引子首先,让我们简单了解一下什么是定时任务。在软件开发中,定时任务就是按照一定的时......
  • C#线程篇---Task(任务)和线程池不得不说的秘密
    C#线程篇---Task(任务)和线程池不得不说的秘密 我们要知道的是,QueueUserWorkItem这个技术存在许多限制。其中最大的问题是没有一个内建的机制让你知道操作在什么时候完成,也没有一个机制在操作完成是获得一个返回值,这些问题使得我们都不敢启用这个技术。Microsoft为了克服......
  • 定时任务及异步,自定义注解进行参数校验
    简单来说:浅拷贝:对基本数据类型进行值传递,对引用数据类型进行引用传递般的拷贝,此为浅拷贝深拷贝:对基本数据类型进行值传递,对引用数据类型,创建一个新的对象,并复制其内容,此为深拷贝。 如何在Spring/SpringBoot中优雅地做参数校验?springboot项目使用validation-api......
  • JUC并发编程 用CompletableFuture 创建异步任务
    1CompletableFuture对Future的改进1.1CompletableFuture为什么会出现get()方法在Future计算完成之前会一直处在阻塞状态下,阻塞的方式和异步编程的设计理念相违背。isDene()方法容易耗费cpu资源(cpu空转),对于真正的异步处理我们希望是可以通过传入回调函数,在Future结束时自动......