Wexflow是一个从纯C#中编写的开源多线程工作流引擎。Wexflow旨在使自动化,工作流程,长期运行的流程以及系统,应用程序和人员之间的交互变得容易,直接和干净。
工作流是一系列不同的步骤或阶段(如上图所示)。每个步骤在Wexflow中作为任务建模。任务可以使用XML在视觉上组合到工作流中。可以通过Wexflow Designer在设计模式下组装任务。此功能位于待办事项列表中,即将开始。
Wexflow提供以下任务:
此时,Wexflow仅支持顺序执行任务,但更复杂的场景(如DoWhile,DoIf,并行任务执行等)都在待办事项列表中,很快就会到来。
在本文中,您将学习如何安装Wexflow,如何卸载它,如何使用它,如何创建自定义任务,最后你会看到它是如何编码的。
在部分使用代码,的源代码版本1.0.1详细解释。源代码自本版本以来已更改,但该部分将为您提供有关Wexflow如何工作的清晰概念。
要使用Wexflow,您需要具备以下基本技能:
要创建自定义任务,您需要具备以下基本技能:
此时,Wexflow仅支持在XML中创建和编辑工作流。但是,在设计模式下创建和编辑工作流程位于待办事项列表中,很快就会完成。Wexflow Designer旨在允许不熟悉XML的人员使用Wexflow,以便他们可以轻松地创建和编辑其工作流程。
Wexflow可以安装在Windows XP,Windows Server 2003及更高版本上。Wexflow支持.NET Framework 4.0及更高版本。
要安装Wexflow,请执行以下操作:
1.安装Microsoft .NET Framework 4.0或更高版本。
2.安装Microsoft Sync Framework 2.1同步可再发行组件(Synchronization-v2.1-x86-ENU.msi在Wexflow_setup.zip中可用)。
3.安装Microsoft同步框架2.1提供服务再分发(ProviderServices-V2.1 86 ENU.MSI提供Wexflow_setup.zip)。
4.安装WexflowSetup.exe(提供Wexflow_setup.zip):
5.您可以选择创建桌面快捷方式:
6.单击安装以执行安装:
7.最后,单击完成以完成安装:
在开始菜单中添加以下菜单:
安装Wexflow后,将安装名为Wexflow的Windows服务并自动启动。要启动Wexflow Manager,此Windows服务必须正在运行。但是,如果你想停止它,你可以从Windows服务控制台:
“文档”菜单打开Wexflow的文档文件夹。“配置”菜单打开Wexflow的配置文件夹。“日志”菜单打开日期的日志文件。
要卸载Wexflow,只需从“Windows开始菜单> Wexflow”中单击“卸载”菜单。
或者进入“配置面板>添加/删除程序”,然后选择“Wexflow版本1.0.1”,然后单击卸载:
之后Wexflow卸载,文件夹C:\Wexflow\
和C:\WexflowTesting\
不会被删除,以防止用户定义的工作流和测试场景被删除。但是,如果您不需要它们,您可以手动删除它们。
日志文件 C:\Program Files\Wexflow\Wexflow.log
也不会被删除跟踪由Wexflow完成最后的操作。但是,如果不需要日志,您可以删除日志文件。
安装Wexflow,文件夹后C:\Wexflow\
和C:\WexflowTesting\
创建。该文件夹C:\Wexflow\
包含以下元素:
Wexflow.xml
这是Wexflow引擎的主配置文件。其路径可以配置自 C:\Program Files\Wexflow\Wexflow.Clients.WindowsService.exe.config
Workflows/
其中包含XML格式的工作流。Temp/
这是Wexflow的临时性。该文件夹C:\WexflowTesting\
中包含的测试场景数据。
日志写入 C:\Program Files\Wexflow\Wexflow.log
。每天有一个日志文件。旧的日志文件保存在这种格式 C:\Program Files\Wexflow\Wexflow.logyyyyMMdd
。
在工作流的配置文件下面:
<?xml version="1.0" encoding="utf-8" ?><!-- This is the configuration file of a workflow. A workflow is composed of: - An id which is an integer that must be unique. - A name which is a string that must be unique. - A description wich is a string. - A settings section which is composed of the following elements: - A launchType which is one of the following options: - startup: The workflow is launched when Wexflow engine starts. - trigger: The workflow is launched manually from the Wexflow manager. - periodic: The workflow is lauched periodically. - A period which is necessary for the periodic launchType. It is a timeSpan in this format dd.hh:mm:ss. For example the period 00.00:02:00 will launch the workflow every 2 minutes. - The enabled option which allows to enable or disable a workflow. The possible values are true or false. - A Tasks section which contains the tasks that will be executed by the workflow one after the other. - A Task is composed of: - An id which is an integer that must be unique. - A name wich is one of the options described in the tasks documentation. - A description which is a string. - The enable option which allows to enable or disable a task. The possible values are true or false. - A collection of settings.--><Workflow id="$int" name="$string" description="$string"> <Settings> <Setting name="launchType" value="startup|trigger|periodic" /> <Setting name="period" value="dd.hh:mm:ss" /> <Setting name="enabled" value="true|false" /> </Settings> <Tasks> <Task id="$int" name="$string" description="$string" enabled="true|false"> <Setting name="$string" value="$string" /> <Setting name="$string" value="$string" /> <!-- You can add as many settings as you want. --> </Task> <Task id="$int" name="$string" description="$string" enabled="true|false"> <Setting name="$string" value="$string" /> <Setting name="$string" value="$string" /> </Task> <!-- You can add as many tasks as you want. --> </Tasks></Workflow>
在name
一个选项Task
必须是下列条件之一:
CsvToXml
:此任务将CSV文件转换为XML文件。输出XML文件的格式在任务的文档中描述。FilesCopier
:此任务将文件集合复制到目标文件夹。 FilesLoader
:此任务加载位于文件夹或通过文件的集合file
选项。 FilesMover
:此任务将文件集合移动到目标文件夹。FilesRemover
:此任务将删除一组文件。Ftp
:此任务允许通过FTP,FTPS(显式/隐式)或SFTP列出,上传,下载或删除文件。这个任务使用用C#编写的开源库。ListEntities
:此任务列出日志中工作流任务加载的所有实体。此任务对于解决问题很有用。ListFiles
:此任务列出日志中工作流任务加载的所有文件。此任务对于解决问题很有用。MailsSender
:此任务从XML文件发送一组电子邮件。任务的文档中描述了输入XML文件的格式。Md5
:此任务生成文件集合的MD5和,并将结果写入XML文件。输出XML文件的格式在任务的文档中描述。Mkdir
:此任务创建文件夹集合。ProcessLauncher
:此任务启动一个进程。如果进程生成一个文件作为输出可以将一个文件集合传递给任务,这样对于每个文件,将通过该进程生成一个输出文件。阅读任务的文档以获取更多信息。Rmdir
:此任务删除一组文件夹。Touch
:此任务创建一个空文件的集合。Twitter
:此任务从XML文件发送一个tweets集合。输入XML文件的格式在任务的文档中描述。XmlToCsv
:此任务将XML文件转换为CSV文件。任务的文档中描述了输入XML文件的格式。Xslt
:此任务转换XML文件。可以使用XSLT 1.0处理器或XSLT 2.0处理器。Zip
:此任务从文件集合创建zip存档。Tar
:此任务从一组文件中创建一个tar存档。Tgz
:此任务从一组文件创建tar.gz归档。Sql
:此任务执行的SQL脚本文件的总汇,或通过一个简单的SQL脚本sql
设置选项。它支持Microsoft SQL Server,Microsoft Access中,甲骨文,MySQL和SQLite的,PostgreSQL和Teradata的。Wmi
:此任务执行WMI查询并在XML文件中输出结果。输出XML文件的格式在任务的文档中描述。ImagesTransformer
:此任务将图像文件的集合转换为指定的格式。输出格式可以是以下之一:Bmp,Emf,Exif,Gif,Icon,Jpeg,Png,Tiff或Wmf。Http
:此任务允许通过HTTP或HTTPS下载文件。Sync
:此任务允许将本地或远程源目录的内容同步到本地或远程目标目录。此任务使用Microsoft Sync Framework 2.1。FilesRenamer
:此任务允许重命名文件系统上的文件集合。该Xslt
任务可用于沿与ListFiles
任务来创建新的文件名。看看这些任务的文档和工作流程示例Workflow_FilesRenamer.xml
,看看如何可以做到这一点。Wait
:此任务等待指定的持续时间。FilesExist
:此任务检查是否存在文件和/或目录的集合。要了解如何使自己的工作流程,你可以在速效检查出的工作流程样本 C:\Wexflow\Workflows\
和阅读的文档文件夹中可用的任务单证C:\Program Files\Wexflow\Documentation
。
如果在创建新的工作流C:\Wexflow\Workflows\
或者如果现有的工作流被删除或修改,则必须重新启动Wexflow Windows服务,使得这些修改生效。
要禁用工作流程,您可以设置enabled
工作流的设置选项false
。如果你想使工作流从Wexflow引擎加载的工作流的列表中消失,您可以创建一个新的目录disabled
中 C:\Wexflow\Workflows\
移动工作流到该目录然后重新启动Wexflow Windows服务。
Wexflow Manager是一个简单的应用程序,允许用户执行以下操作:
看到发生了什么事情在Wexflow,打开日志文件 C:\Program Files\Wexflow\Wexflow.log
在诸如文本编辑器记事本+ +。 Notepad ++会在日志文件填满时更新日志文件。
在这一节中,很少有工作流样品将为了使熟悉Wexflow工作流synthax终端用户呈现。
此工作流将发票上传到SFTP服务器,然后等待2天,然后通知客户。
<Workflow id="99" name="Workflow_Invoices" description="Workflow_Invoices"> <Settings> <Setting name="launchType" value="trigger" /> <Setting name="enabled" value="true" /> </Settings> <Tasks> <Task id="1" name="FilesLoader" description="Loading invioces" enabled="true"> <Setting name="folder" value="C:\WexflowTesting\Invoices\" /> </Task> <Task id="2" name="Ftp" description="Uploading invoices" enabled="true"> <Setting name="protocol" value="sftp" /> <!-- ftp|ftps|sftp --> <Setting name="command" value="upload" /> <!-- list|upload|download|delete --> <Setting name="server" value="127.0.1" /> <Setting name="port" value="21" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> <Setting name="path" value="/" /> <Setting name="selectFiles" value="1" /> </Task> <Task id="3" name="Wait" description="Waiting for 2 days" enabled="true"> <Setting name="duration" value="2.00:00:00" /> </Task> <Task id="4" name="FilesLoader" description="Loading emails" enabled="true"> <Setting name="file" value="C:\WexflowTesting\Emails\Invoices.xml" /> </Task> <Task id="5" name="MailsSender" description="Notifying customers" enabled="true"> <Setting name="selectFiles" value="4" /> <Setting name="host" value="127.0.0.1" /> <Setting name="port" value="587" /> <Setting name="enableSsl" value="true" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> </Task> <Task id="6" name="FilesMover" description="Moving invoices" enabled="true"> <Setting name="selectFiles" value="1" /> <Setting name="destFolder" value="C:\WexflowTesting\Invoices_sent\" /> </Task> </Tasks></Workflow>
首先,在FilesLoader
所在的文件夹中的任务加载所有的发票 C:\WexflowTesting\Invoices\
,然后将Ftp
任务它们上传到SFTP服务器,那么Wait
任务等待2天,然后将FilesLoader
任务加载邮件以XML格式,然后将MailsSender
任务发送电子邮件。最后,FilesMover
任务会将发票到该文件夹 C:\WexflowTesting\Invoices_sent\
。
该工作流等待文件在抵达C:\WexflowTesting\Watchfolder1\
和 C:\WexflowTesting\Watchfolder2\
然后将它们上传到FTP服务器,然后将它们移动到C:\WexflowTesting\Sent\
文件夹中。此工作流程每2分钟启动一次。
<Workflow id="6" name="Workflow_FilesSender" description="Workflow_FilesSender"> <Settings> <Setting name="launchType" value="periodic" /> <Setting name="period" value="00.00:02:00.00" /> <Setting name="enabled" value="true" /> </Settings> <Tasks> <Task id="1" name="FilesLoader" description="Loading files" enabled="true"> <Setting name="folder" value="C:\WexflowTesting\Watchfolder1\" /> <Setting name="folder" value="C:\WexflowTesting\Watchfolder2\" /> </Task> <Task id="2" name="Ftp" description="Uploading files" enabled="true"> <Setting name="protocol" value="ftp" /> <!-- ftp|ftps|sftp --> <Setting name="command" value="upload" /> <!-- list|upload|download|delete --> <Setting name="server" value="127.0.1" /> <Setting name="port" value="21" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> <Setting name="path" value="/" /> <Setting name="selectFiles" value="1" /> </Task> <Task id="3" name="FilesMover" description="Moving files to Sent folder" enabled="true"> <Setting name="selectFiles" value="1" /> <Setting name="destFolder" value="C:\WexflowTesting\Sent\" /> </Task> </Tasks></Workflow>
首先,在FilesLoader
任务负载分布在文件夹中所有的文件 C:\WexflowTesting\Watchfolder1\
和 C:\WexflowTesting\Watchfolder2\
随后的Ftp
任务加载文件,并将其上传到FTP服务器。最后,该FilesMover
任务中的文件移动到文件夹中 C:\WexflowTesting\Sent\
。
此工作流程转码位于WAV文件C:\WexflowTesting\WAV\
为MP3格式通过FFMPEG和转换的文件移动到 C:\WexflowTesting\MP3\
。
<Workflow id="12" name="Workflow_ffmpeg" description="Workflow_ffmpeg"> <Settings> <Setting name="launchType" value="trigger" /> <Setting name="enabled" value="true" /> </Settings> <Tasks> <Task id="1" name="FilesLoader" description="Loading WAV files" enabled="true"> <Setting name="folder" value="C:\WexflowTesting\WAV\" /> </Task> <Task id="2" name="ProcessLauncher" description="WAV to MP3" enabled="true"> <Setting name="selectFiles" value="1" /> <!-- You need to install FFMPEG --> <Setting name="processPath" value="C:\Program Files\ffmpeg\bin\ffmpeg.exe" /> <!-- variables: {$filePath},{$fileName},{$fileNameWithoutExtension}--> <Setting name="processCmd" value="-i {$filePath} -codec:a libmp3lame -qscale:a 2 {$output:$fileNameWithoutExtension.mp3}" /> <Setting name="hideGui" value="true" /> <Setting name="generatesFiles" value="true" /> </Task> <Task id="3" name="FilesMover" description="Moving MP3 files from temp folder" enabled="true"> <Setting name="selectFiles" value="2" /> <Setting name="destFolder" value="C:\WexflowTesting\MP3\" /> </Task> </Tasks></Workflow>
首先,在FilesLoader
任务负载所在的文件夹中的所有文件 C:\WexflowTesting\WAV\
,然后将 ProcessLauncher
任务按以创建MP3文件中指定正确的命令启动上的每个文件FFMPEG过程。最后,该FilesMover
任务的MP3文件移动到文件夹中 C:\WexflowTesting\MP3\
。
该工作流等待WAV文件到达C:\WexflowTesting\WAV\
然后将它们转码为MP3文件通过VLC然后上传MP3文件到FTP服务器,然后移动WAV文件 C:\WexflowTesting\WAV_processed\
。此工作流程每2分钟启动一次。
<Workflow id="13" name="Workflow_vlc" description="Workflow_vlc"> <Settings> <Setting name="launchType" value="periodic" /> <Setting name="period" value="00.00:02:00.00" /> <Setting name="enabled" value="true" /> </Settings> <Tasks> <Task id="1" name="FilesLoader" description="Loading WAV files" enabled="true"> <Setting name="folder" value="C:\WexflowTesting\WAV\" /> </Task> <Task id="2" name="ProcessLauncher" description="WAV to MP3" enabled="true"> <Setting name="selectFiles" value="1" /> <!-- You need to install VLC--> <Setting name="processPath" value="C:\Program Files\VideoLAN\VLC\vlc.exe" /> <!-- variables: {$filePath},{$fileName},{$fileNameWithoutExtension}--> <Setting name="processCmd" value="-I dummy {$filePath} :sout=#transcode{acodec=mpga}:std{dst={$output:$fileNameWithoutExtension.mp3},access=file} vlc://quit" /> <Setting name="hideGui" value="true" /> <Setting name="generatesFiles" value="true" /> </Task> <Task id="3" name="Ftp" description="Uploading MP3 files" enabled="true"> <Setting name="protocol" value="ftp" /> <Setting name="command" value="upload" /> <Setting name="server" value="127.0.1" /> <Setting name="port" value="21" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> <Setting name="path" value="/" /> <Setting name="selectFiles" value="2" /> </Task> <Task id="4" name="FilesMover" description="Moving WAV files" enabled="true"> <Setting name="selectFiles" value="1" /> <Setting name="destFolder" value="C:\WexflowTesting\WAV_processed\" /> </Task> </Tasks></Workflow>
首先,在FilesLoader
任务负载所在的文件夹中的所有文件 C:\WexflowTesting\WAV\
,然后将 ProcessLauncher
任务按以创建MP3文件中指定正确的命令启动上的每个文件VLC过程。然后,Ftp
任务加载由产生的MP3文件 ProcessLauncher
的任务,然后将它们上传到FTP服务器。最后,该FilesMover
任务的处理的WAV文件移动到文件夹中 C:\WexflowTesting\WAV_processed\
。
此工作流从FTP服务器下载特定文件。此工作流通过列出位于服务器的根文件夹中的所有文件启动,那么将被下载通过XSLT(被标记的特定文件LisFiles.xslt
),然后将这些文件被下载的Ftp
任务,通过 todo="toDownload"
和from="app4"
标签,然后将下载的文件移动到文件夹中C:\WexflowTesting\Ftp_download\
。
<Workflow id="40" name="Workflow_Ftp_download_tag" description="Workflow_Ftp_download_tag"> <Settings> <Setting name="launchType" value="trigger" /> <!-- startup|trigger|periodic --> <Setting name="enabled" value="true" /> <!-- true|false --> </Settings> <Tasks> <Task id="1" name="Ftp" description="Listing files (FTP)" enabled="true"> <Setting name="command" value="list" /> <Setting name="protocol" value="ftp" /> <!-- ftp|ftps|sftp --> <Setting name="server" value="127.0.1" /> <Setting name="port" value="21" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> <Setting name="path" value="/" /> </Task> <Task id="2" name="ListFiles" description="Listing files" enabled="true"> </Task> <Task id="3" name="Xslt" description="Renaming and tagging files" enabled="true"> <Setting name="selectFiles" value="2" /> <Setting name="xsltPath" value="C:\Wexflow\Xslt\ListFiles.xslt" /> <Setting name="version" value="2.0" /> <!-- 1.0|2.0 --> <Setting name="removeWexflowProcessingNodes" value="false" /> </Task> <Task id="4" name="Ftp" description="Downloading files" enabled="true"> <Setting name="command" value="download" /> <Setting name="protocol" value="ftp" /> <!-- ftp|ftps|sftp --> <Setting name="server" value="127.0.1" /> <Setting name="port" value="21" /> <Setting name="user" value="user" /> <Setting name="password" value="password" /> <Setting name="path" value="/" /> <Setting name="selectFiles" todo="toDownload" from="app4" /> </Task> <Task id="5" name="FilesMover" description="Moving files to Ftp_download" enabled="true"> <Setting name="selectFiles" value="4" /> <Setting name="destFolder" value="C:\WexflowTesting\Ftp_download\" /> <Setting name="overwrite" value="true" /> </Task> </Tasks></Workflow>
粗略地说,该Ftp
任务会加载位于FTP服务器的根文件夹中的工作流的运行实例的文件列表,那么ListFiles
包含加载的所有文件,任务输出和XML文件,然后将Xslt
任务作为输入这个XML并生成一个XML至极包含一个名为系统节点<WexflowProcessing>
至极包含的文件列表进行标记和/或重命名。
为了理解标记和重命名文件的工作方式,请参阅的文档ListFiles
和Xslt
任务。
下面是对XSLT ListFiles.xslt
用于标记文件:
<?xml version="1.0" encoding="utf-8"?><xsl:stylesheet version="1.0" xmlns:xsl="http://www.w3.org/1999/XSL/Transform"> <xsl:output method="xml" indent="yes"/> <xsl:template match="/"> <root> <WexflowProcessing> <xsl:for-each select="//WexflowProcessing/Workflow/Files//File"> <xsl:choose> <xsl:when test="@name = 'file1.txt'"> <File taskId="{@taskId}" name="{@name}" renameTo="file1_renamed.txt" todo="toRename" from="app1" /> </xsl:when> <xsl:when test="@name = 'file2.txt'"> <File taskId="{@taskId}" name="{@name}" renameTo="file2_renamed.txt" todo="toSend" from="app2" /> </xsl:when> <xsl:when test="@name = 'file3.txt'"> <File taskId="{@taskId}" name="{@name}" renameTo="file3_renamed.txt" todo="toDownload" from="app3" /> </xsl:when> <xsl:when test="@name = 'file4.txt'"> <File taskId="{@taskId}" name="{@name}" renameTo="file4_renamed.txt" todo="toDownload" from="app4" /> </xsl:when> </xsl:choose> </xsl:for-each> </WexflowProcessing> </root> </xsl:template></xsl:stylesheet>
这些是简单和基本的工作流程,以了解如何制作自己的工作流程。但是,如果您有多个系统,应用和参与的工作流程自动化,工作流可能是非常有趣的。
自定义任务是工作流引擎中必须的,允许系统和应用程序进行交互。
要创建自定义任务,MyTask
比如你需要进行如下操作:
Wexflow.Tasks.MyTask
。Wexflow.Core.dll
和log4net.dll
。这些组件都位于Wexflow的安装文件夹C:\Program Files\Wexflow\
。MyTask
实现抽象类 Wexflow.Core.Task
。Wexflow.Tasks.MyTask
代码应如下所示:
public class MyTask : Wexflow.Core.Task{ public MyTask(XElement xe, Workflow wf) : base(xe, wf) { // Task settings goes here } public override void Run() { try { // Task logic goes here } catch (ThreadAbortException) { throw; } }}
要检索设置,可以使用以下方法:
string settingValue = this.GetSetting("settingName");string settingValue = this.GetSetting("settingName", defaultValue);string[] settingValues = this.GetSettings("settingName");
要选择通过在工作流的运行实例加载的文件selectFiles
设置选项,你可以如下做到这一点:
FileInf[] files = this.SelectFiles();
要选择通过在工作流的运行实例加载实体selectEntities
的设置选项,你可以如下做到这一点:
Entity[] entities = this.SelectEntities();
在Entity
与操纵从数据库或Web服务,例如对象的自定义任务时类可以是非常有用的。
要在任务中加载文件,可以执行如下操作:
this.Files.Add(new FileInf(path, this.Id));
要在任务中加载实体,您可以执行如下操作:
this.Entities.Add(myEntity);
最后,如果你编码完你的自定义任务,编译类库项目和复制装配Wexflow.Tasks.MyTask.dll
在 C:\Program Files\Wexflow\
。您的自定义任务随后可以使用如下:
<Task id="$int" name="MyTask" description="My task description" enabled="true"> <Setting name="settingName" value="settingValue" /></Task>
而已。这是所有你需要知道的事情,开始编码自己的自定义任务。
要调试Wexflow,请执行以下操作:
要调试Wexflow Windows服务(Wexflow.Clients.WindowsService项目),请在“项目设置>调试>启动选项”中添加“debug”命令行参数:
要调试Wexflow管理器(Wexflow.Clients.Manager项目),在“Propject设置>调试>启动选项”添加“调试”命令行参数:
在本节中,的源代码版本1.0.1进行详细的说明。源代码自本版本以来已更改,但本节将为您提供一个清晰的Wexflow如何工作原理。
Wexflow的源代码很容易理解。源代码项目组织如下:
该Wexflow.Clients
解决方案文件夹中包含Wexflow Windows服务,Wexflow经理。该Wexflow.Core
解决方案文件夹包含Wexflow的核心组件Wexflow.Core
,在WCF服务合同Wexflow.Core.Service.Contracts
,并在WCF代理Wexflow.Core.Service.Client
。该Wexflow.Tasks
包含Wexflow任务。
Wexflow使用承载WCF Web服务的Windows服务。当Wexflow Windows服务启动的新实例WexflowEngine
被创建然后通过方法拼命地跑Run()
。WCF Web服务允许执行以下操作:
WexflowEngine
。WorkflowInfo
从工作流ID对象。以下Wexflow Windows服务的源代码。
public partial class WexflowWindowsService : ServiceBase{ public static string SETTINGS_FILE = ConfigurationManager.AppSettings["WexflowSettingsFile"]; public static WexflowEngine WEXFLOW_ENGINE = new WexflowEngine(SETTINGS_FILE); private ServiceHost _serviceHost = null; public WexflowWindowsService() { InitializeComponent(); this.ServiceName = "Wexflow"; WEXFLOW_ENGINE.Run(); } public void OnDebug() { this.OnStart(null); } protected override void OnStart(string[] args) { if (this._serviceHost != null) { this._serviceHost.Close(); } // Create a ServiceHost for the WexflowService type and // provide the base address. this._serviceHost = new ServiceHost(typeof(WexflowService)); // Open the ServiceHostBase to create listeners and start // listening for messages. this._serviceHost.Open(); } protected override void OnStop() { if (this._serviceHost != null) { this._serviceHost.Close(); this._serviceHost = null; } }}
WexflowSettingsFile
默认为 C:\Wexflow\Wexflow.xml
。
ServiceHost
用于承载WexflowService
在Wexflow Windows服务WCF Web服务。
下面的源代码 WexflowService
。
[ServiceContract(Namespace = "http://WexflowService/")]public interface IWexflowService{ [OperationContract] string Hello(); [OperationContract] WorkflowInfo[] GetWorkflows(); [OperationContract] void StartWorkflow(int workflowId); [OperationContract] void StopWorkflow(int workflowId); [OperationContract] void SuspendWorkflow(int workflowId); [OperationContract] void ResumeWorkflow(int workflowId); [OperationContract] WorkflowInfo GetWorkflow(int workflowId);}[ServiceBehavior(IncludeExceptionDetailInFaults=true)]public class WexflowService:IWexflowService{ public string Hello() { return "Hello!"; } public WorkflowInfo[] GetWorkflows() { List<WorkflowInfo> wfis = new List<WorkflowInfo>(); foreach (Workflow wf in WexflowWindowsService.WEXFLOW_ENGINE.Workflows) { wfis.Add(new WorkflowInfo(wf.Id, wf.Name, wf.LaunchType, wf.IsEnabled, wf.Description, wf.IsRunning, wf.IsPaused)); } return wfis.ToArray(); } public void StartWorkflow(int workflowId) { WexflowWindowsService.WEXFLOW_ENGINE.StartWorkflow(workflowId); } public void StopWorkflow(int workflowId) { WexflowWindowsService.WEXFLOW_ENGINE.StopWorkflow(workflowId); } public void SuspendWorkflow(int workflowId) { WexflowWindowsService.WEXFLOW_ENGINE.PauseWorkflow(workflowId); } public void ResumeWorkflow(int workflowId) { WexflowWindowsService.WEXFLOW_ENGINE.ResumeWorkflow(workflowId); } public WorkflowInfo GetWorkflow(int workflowId) { Workflow wf = WexflowWindowsService.WEXFLOW_ENGINE.GetWorkflow(workflowId); return new WorkflowInfo(wf.Id, wf.Name, wf.LaunchType, wf.IsEnabled, wf.Description, wf.IsRunning, wf.IsPaused); }}
下面system.serviceModel
Wexflow Windows服务的配置。
<system.serviceModel> <services> <service behaviorConfiguration="WexflowServiceBehavior" name="Wexflow.Clients.WindowsService.WexflowService"> <endpoint address="" binding="wsHttpBinding" contract="Wexflow.Clients.WindowsService.IWexflowService" /> <endpoint address="mex" binding="mexHttpBinding" contract="IMetadataExchange" /> <host> <baseAddresses> <add baseAddress="http://localhost:8000/WexflowService/" /> </baseAddresses> </host> </service> </services> <behaviors> <serviceBehaviors> <behavior name="WexflowServiceBehavior"> <serviceMetadata httpGetEnabled="true"/> <serviceDebug includeExceptionDetailInFaults="False"/> </behavior> </serviceBehaviors> </behaviors> </system.serviceModel>
SvcUtil.exe
用于生成WCF代理如下:
SvcUtil.exe http://localhost:8000/WexflowService/ /out:WexflowServiceClient.cs /config:App.config
要调试Wexflow Windows服务,使用了以下技巧:
namespace Wexflow.Clients.WindowsService{ static class Program { static void Main(string[] args) { if (args.Length > 0 && args[0].Equals("debug")) { WexflowWindowsService service = new WexflowWindowsService(); service.OnDebug(); Thread.Sleep(Timeout.Infinite); } else { ServiceBase[] servicesToRun = new ServiceBase[] { new WexflowWindowsService() }; ServiceBase.Run(servicesToRun); } } }}
下面的类图描述了Wexflow Engine的一般体系结构。
WexflowEngine
类由工作流和其他属性的集合组成。工作流由一组任务和其他属性组成。Task
是Wexflow任务继承的抽象类。
粗略地说,WexflowEngine
类解析工作流的所有XML文件转换成Workflow
通过LINQ对象到XML。然后,当该方法Run()
这个类被调用时,具有工作流startup
作为launchType
自动启动,则具有与工作流periodic
作为launchType
通过一个定时器被启动。
当一个新的实例WexflowEngine
被创建以下的事情发生:
public WexflowEngine(string settingsFile) { this.SettingsFile = settingsFile; LoadSettings(); LoadWorkflows();}
加载工作流引擎设置,然后从其XML文件加载工作流。
加载设置如下:
private void LoadSettings(){ XDocument xdoc = XDocument.Load(this.SettingsFile); this.WorkflowsFolder = GetWexflowSetting(xdoc, "workflowsFolder"); this.TempFolder = GetWexflowSetting(xdoc, "tempFolder");}private string GetWexflowSetting(XDocument xdoc, string name){ return xdoc.XPathSelectElement(string.Format("/Wexflow/Setting[@name='{0}']", name)).Attribute("value").Value;}
加载工作流程如下:
private void LoadWorkflows(){ List<Workflow> workflows = new List<Workflow>(); foreach (string file in Directory.GetFiles(this.WorkflowsFolder)) { try { Workflow workflow = new Workflow(file, this.TempFolder); workflows.Add(workflow); Logger.InfoFormat("Workflow loaded: {0}", workflow); } catch (Exception e) { Logger.ErrorFormat("An error occured while loading the workflow : {0} Please check the workflow configuration.", file); } } this.Workflows = workflows.ToArray();}
当创建工作流的新实例时,会发生以下事件:
public Workflow(string path, string wexflowTempFolder){ this.JobId = 1; this._thread = null; this.WorkflowFilePath = path; this.WexflowTempFolder = wexflowTempFolder; this.FilesPerTask = new Dictionary<int, List<FileInf>>(); this.EntitiesPerTask = new Dictionary<int, List<Entity>>(); Load();}private void Load(){ XDocument xdoc = XDocument.Load(this.WorkflowFilePath); this.Id = int.Parse(GetWorkflowAttribute(xdoc, "id")); this.Name = GetWorkflowAttribute(xdoc, "name"); this.Description = GetWorkflowAttribute(xdoc, "description"); this.LaunchType = (LaunchType)Enum.Parse(typeof(LaunchType), GetWorkflowSetting(xdoc, "launchType"), true); if(this.LaunchType == Core.LaunchType.Periodic) this.Period = TimeSpan.Parse(GetWorkflowSetting(xdoc, "period")); this.IsEnabled = bool.Parse(GetWorkflowSetting(xdoc, "enabled")); List<Task> tasks = new List<Task>(); foreach (XElement xTask in xdoc.XPathSelectElements("/Workflow/Tasks/Task")) { string name = xTask.Attribute("name").Value; string assemblyName = "Wexflow.Tasks." + name; string typeName = "Wexflow.Tasks." + name + "." + name + ", " + assemblyName; Task task = (Task)Activator.CreateInstance(Type.GetType(typeName), xTask, this); tasks.Add(task); } this.Taks = tasks.ToArray();}private string GetWorkflowAttribute(XDocument xdoc, string attr){ return xdoc.XPathSelectElement("/Workflow").Attribute(attr).Value;}private string GetWorkflowSetting(XDocument xdoc, string name){ return xdoc.XPathSelectElement(string.Format("/Workflow[@id='{0}']/Settings/Setting[@name='{1}']", this.Id, name)).Attribute("value").Value;}
首先,检索工作流设置,然后通过反射构建任务。
当Wexflow引擎开始通过该方法Run()
如下事情发生:
public void Run(){ foreach (Workflow workflow in this.Workflows) { if (workflow.IsEnabled) { if (workflow.LaunchType == LaunchType.Startup) { workflow.Start(); } else if (workflow.LaunchType == LaunchType.Periodic) { Action<object> callback = o => { Workflow wf = (Workflow)o; if (!wf.IsRunning) wf.Start(); }; WexflowTimer timer = new WexflowTimer(new TimerCallback(callback), workflow, workflow.Period); timer.Start(); } } }}
对于每一个工作流程,如果启用了工作流程,Wexflow将启动它如果launchType
是startup
,否则如果launchType
是periodic
Wexflow将创建一个WexflowTimer
将启动工作流程的每个period
。
的WexflowTimer
类是很简单的,看起来像如下:
public class WexflowTimer{ public TimerCallback TimerCallback { get; private set; } public object State { get; private set; } public TimeSpan Period { get; private set; } public WexflowTimer(TimerCallback timerCallback, object state, TimeSpan period) { this.TimerCallback = timerCallback; this.State = state; this.Period = period; } public void Start() { Thread thread = new Thread(new ThreadStart(() => { Stopwatch stopwatch = new Stopwatch(); stopwatch.Start(); for(;;) { if (stopwatch.ElapsedMilliseconds >= this.Period.TotalMilliseconds) { stopwatch.Reset(); stopwatch.Start(); this.TimerCallback.Invoke(this.State); } Thread.Sleep(100); } })); thread.Start(); }}
我用一个自定义的计时器,因为我遇到了很多问题蒙山System.Threading.Timer
同时测试复杂的周期性工作流程。
当工作流启动时,会发生以下情况:
public void Start(){ Thread thread = new Thread(new ThreadStart(() => { try { this.IsRunning = true; Logger.InfoFormat("{0} Workflow started.", this.LogTag); // Create the temp folder CreateTempFolder(); // Run the tasks foreach (Task task in this.Taks) { if (task.IsEnabled) { task.Run(); } } } catch (ThreadAbortException) { } catch (Exception e) { Logger.ErrorFormat("An error occured while running the workflow : {0}", e, this); } finally { // Cleanup foreach (List<FileInf> files in this.FilesPerTask.Values) files.Clear(); foreach (List<Entity> entities in this.EntitiesPerTask.Values) entities.Clear(); this._thread = null; this.IsRunning = false; GC.Collect(); Logger.InfoFormat("{0} Workflow finished.", this.LogTag); this.JobId++; } })); this._thread = thread; thread.Start();}private void CreateTempFolder(){ // WorkflowId/dd-MM-yyyy/HH-mm-ss-fff string wfTempFolder = Path.Combine(this.WexflowTempFolder, this.Id.ToString()); if (!Directory.Exists(wfTempFolder)) Directory.CreateDirectory(wfTempFolder); string wfDayTempFolder = Path.Combine(wfTempFolder, string.Format("{0:yyyy-MM-dd}", DateTime.Now)); if (!Directory.Exists(wfDayTempFolder)) Directory.CreateDirectory(wfDayTempFolder); string wfJobTempFolder = Path.Combine(wfDayTempFolder, string.Format("{0:HH-mm-ss-fff}", DateTime.Now)); if (!Directory.Exists(wfJobTempFolder)) Directory.CreateDirectory(wfJobTempFolder); this.WorkflowTempFolder = wfJobTempFolder;}
创建并启动一个新线程。在此线程中,创建一个临时文件夹,然后逐个启动任务。在工作流结束时,将清除由任务加载的文件和实体。
下面的FileInf
类:
public class FileInf{ public string Path { get; private set; } public string FileName { get; private set; } public int TaskId { get; private set; } public FileInf(string path, int taskId) { this.Path = path; this.FileName = System.IO.Path.GetFileName(this.Path); this.TaskId = taskId; }}
下面的Entity
类:
public abstract class Entity{ public int TaskId { get; private set; }}
其工作流程trigger
是launchType
通过调用从Wexflow Manager启动的Workflow.Start()
方法。
可以从Wexflow Manager停止,挂起和恢复工作流。
最后,每个任务实现抽象类Wexflow.Core.Task
:
public abstract class Task{ public int Id { get; private set; } public string Name { get; private set; } public string Description { get; private set; } public bool IsEnabled { get; private set; } public Workflow Workflow { get; private set; } public List<FileInf> Files { get { return this.Workflow.FilesPerTask[this.Id]; } } public List<Entity> Entities { get { return this.Workflow.EntitiesPerTask[this.Id]; } } private XElement _xElement; public Task(XElement xe, Workflow wf) { this._xElement = xe; this.Id = int.Parse(xe.Attribute("id").Value); this.Name = xe.Attribute("name").Value; this.Description = xe.Attribute("description").Value; this.IsEnabled = bool.Parse(xe.Attribute("enabled").Value); this.Workflow = wf; this.Workflow.FilesPerTask.Add(this.Id, new List<FileInf>()); this.Workflow.EntitiesPerTask.Add(this.Id, new List<Entity>()); } public abstract void Run(); public string GetSetting(string name) { return this._xElement.XPathSelectElement(string.Format("Setting[@name='{0}']", name)).Attribute("value").Value; } public string GetSetting(string name, string defaultValue) { XElement xe = this._xElement.XPathSelectElement(string.Format("Setting[@name='{0}']", name)); if (xe == null) return defaultValue; return xe.Attribute("value").Value; } public string[] GetSettings(string name) { List<string> settings = new List<string>(); foreach (XElement xe in this._xElement.XPathSelectElements(string.Format("Setting[@name='{0}']", name))) { settings.Add(xe.Attribute("value").Value); } return settings.ToArray(); } public FileInf[] SelectFiles() { List<FileInf> files = new List<FileInf>(); foreach (string id in this.GetSettings("selectFiles")) { int taskId = int.Parse(id); files.AddRange(this.Workflow.FilesPerTask[taskId]); } return files.ToArray(); } public Entity[] SelectEntities() { List<Entity> entities = new List<Entity>(); foreach (string id in this.GetSettings("selectEntities")) { int taskId = int.Parse(id); entities.AddRange(this.Workflow.EntitiesPerTask[taskId]); } return entities.ToArray(); } // ...}
每个任务都实现了一个自定义逻辑Run()
方法。每个任务可以加载通过文件的集合Files
属性,并通过实体的集合Entities
属性。每个任务的源代码将不会在本文中详细介绍,但您可以查看每个任务的源代码,以便您可以看到如何构建自定义任务。
Wexflow Manager是一个非常简单的窗体应用程序,它显示了Wexflow加载的所有工作流,并允许最终用户启动,停止,暂停或恢复工作流。
Wexflow管理器的源代码如下所示:
private const string COLUMN_ID = "Id";private const string COLUMN_ENABLED = "Enabled";private const int TIMER_INTERVAL = 100; // msprivate WexflowServiceClient _wexflowServiceClient;private WorkflowInfo[] _workflows;private Dictionary<int, Timer> _timers;private Dictionary<int, bool> _previousIsRunning;private Dictionary<int, bool> _previousIsPaused;private bool _windowsServiceWasStopped;public Form1(){ InitializeComponent(); this.textBoxInfo.Text = "Loading workflows..."; this._timers = new Dictionary<int, Timer>(); this._previousIsRunning = new Dictionary<int, bool>(); this._previousIsPaused = new Dictionary<int, bool>(); this.backgroundWorker1.RunWorkerAsync();}private void backgroundWorker1_DoWork(object sender, DoWorkEventArgs e){ if (Program.DEBUG_MODE || Program.IsWexflowWindowsServiceRunning()) { this._wexflowServiceClient = new WexflowServiceClient(); this._workflows = _wexflowServiceClient.GetWorkflows(); } else { this._workflows = new WorkflowInfo[] { }; this.textBoxInfo.Text = ""; }}private void backgroundWorker1_RunWorkerCompleted(object sender, RunWorkerCompletedEventArgs e){ BindDataGridView(); }private void BindDataGridView(){ SortableBindingList<WorkflowDataInfo> workflows = new SortableBindingList<WorkflowDataInfo>(); foreach (WorkflowInfo workflow in this._workflows) { workflows.Add(new WorkflowDataInfo(workflow.Id, workflow.Name, workflow.LaunchType, workflow.IsEnabled, workflow.Description)); } this.dataGridViewWorkflows.DataSource = workflows; this.dataGridViewWorkflows.Columns[0].AutoSizeMode = DataGridViewAutoSizeColumnMode.DisplayedCells; this.dataGridViewWorkflows.Columns[1].AutoSizeMode = DataGridViewAutoSizeColumnMode.DisplayedCells; this.dataGridViewWorkflows.Columns[2].AutoSizeMode = DataGridViewAutoSizeColumnMode.DisplayedCells; this.dataGridViewWorkflows.Columns[3].AutoSizeMode = DataGridViewAutoSizeColumnMode.DisplayedCells; this.dataGridViewWorkflows.Columns[3].Name = COLUMN_ENABLED; this.dataGridViewWorkflows.Columns[3].HeaderText = COLUMN_ENABLED; this.dataGridViewWorkflows.Columns[4].AutoSizeMode = DataGridViewAutoSizeColumnMode.Fill; this.dataGridViewWorkflows.Sort(this.dataGridViewWorkflows.Columns[0], ListSortDirection.Ascending);}
当的新实例 Form1
被创建一个BackgroundWorker
异步启动。这种BackgroundWorker
检索由Wexflow引擎加载的工作流程和绑定DataGridview
的OnCompleted
事件。
下面的WorkflowDataInfo
类:
public class WorkflowDataInfo:IComparable{ public int Id { get; private set; } public string Name { get; private set; } public LaunchType LaunchType { get; private set; } public bool IsEnabled { get; private set; } public string Description { get; private set; } public WorkflowDataInfo(int id, string name, LaunchType launchType, bool isEnabled, string desc) { this.Id = id; this.Name = name; this.LaunchType = launchType; this.IsEnabled = isEnabled; this.Description = desc; } public int CompareTo(object obj) { WorkflowDataInfo wf = (WorkflowDataInfo)obj; return wf.Id.CompareTo(this.Id); }}
下面的实施Click
各按钮的事件:
private void buttonStart_Click(object sender, EventArgs e){ int wfId = GetSlectedWorkflowId(); if (wfId > -1) { this._wexflowServiceClient.StartWorkflow(wfId); }}private void buttonPause_Click(object sender, EventArgs e){ int wfId = GetSlectedWorkflowId(); if (wfId > -1) { this._wexflowServiceClient.SuspendWorkflow(wfId); this.UpdateButtons(wfId, true); }}private void buttonResume_Click(object sender, EventArgs e){ int wfId = GetSlectedWorkflowId(); if (wfId > -1) { this._wexflowServiceClient.ResumeWorkflow(wfId); }}private void buttonStop_Click(object sender, EventArgs e){ int wfId = GetSlectedWorkflowId(); if (wfId > -1) { this._wexflowServiceClient.StopWorkflow(wfId); this.UpdateButtons(wfId, true); }}private int GetSlectedWorkflowId(){ int wfId = -1; if (dataGridViewWorkflows.SelectedRows.Count > 0) { if(Program.DEBUG_MODE || Program.IsWexflowWindowsServiceRunning()) { wfId = int.Parse(dataGridViewWorkflows.SelectedRows[0].Cells[COLUMN_ID].Value.ToString()); } else { HandleNonRunningWindowsService(); } } return wfId;}private WorkflowInfo GetWorkflow(int id){ if (Program.DEBUG_MODE || Program.IsWexflowWindowsServiceRunning()) { if (this._windowsServiceWasStopped) { this._wexflowServiceClient = new WexflowServiceClient(); this._windowsServiceWasStopped = false; this.backgroundWorker1.RunWorkerAsync(); } return this._wexflowServiceClient.GetWorkflow(id); } else { this._windowsServiceWasStopped = true; HandleNonRunningWindowsService(); } return null;}private void HandleNonRunningWindowsService(){ this.buttonStart.Enabled = this.buttonPause.Enabled = this.buttonResume.Enabled = this.buttonStop.Enabled = false; this.textBoxInfo.Text = "Wexflow Windows Service is not running.";}private void UpdateButtons(int wfId, bool force){ if (wfId > -1) { Workflow workflow = _wexflowEngine.GetWorkflow(wfId); if (workflow != null) { if (!workflow.IsEnabled) { this.textBoxInfo.Text = "This workflow is disabled."; this.buttonStart.Enabled = this.buttonPause.Enabled = this.buttonResume.Enabled = this.buttonStop.Enabled = false; } else { if (!force && !WorkflowStatusChanged(workflow)) return; buttonStart.Enabled = !workflow.IsRunning; buttonStop.Enabled = workflow.IsRunning && !workflow.IsPaused; buttonPause.Enabled = workflow.IsRunning && !workflow.IsPaused; buttonResume.Enabled = workflow.IsPaused; if (workflow.IsRunning && !workflow.IsPaused) { this.textBoxInfo.Text = "This workflow is running..."; } else if (workflow.IsPaused) { this.textBoxInfo.Text = "This workflow is paused."; } else { this.textBoxInfo.Text = ""; } } } }}private bool WorkflowStatusChanged(Workflow workflow){ bool changed = false; if (!this._previousIsRunning.ContainsKey(workflow.Id)) { this._previousIsRunning.Add(workflow.Id, workflow.IsRunning); changed = true; } if (!this._previousIsPaused.ContainsKey(workflow.Id)) { this._previousIsPaused.Add(workflow.Id, workflow.IsPaused); changed = true; } if (changed) { return true; } else { if (this._previousIsRunning[workflow.Id] != workflow.IsRunning) { changed = true; } if (this._previousIsPaused[workflow.Id] != workflow.IsPaused) { changed = true; } this._previousIsRunning[workflow.Id] = workflow.IsRunning; this._previousIsPaused[workflow.Id] = workflow.IsPaused; return changed; }}
下面执行SelectionChanged
的情况DataGridView
:
private void dataGridViewWorkflows_SelectionChanged(object sender, EventArgs e){ int wfId = GetSlectedWorkflowId(); if (wfId > -1) { Workflow workflow = _wexflowEngine.GetWorkflow(wfId); foreach (Timer timer in this._timers.Values) timer.Stop(); if (workflow.IsEnabled) { if (!this._timers.ContainsKey(wfId)) { Timer timer = new Timer(); timer.Interval = TIMER_INTERVAL; timer.Tick += new EventHandler((o, ea) => { this.UpdateButtons(wfId, false); }); this._timers.Add(wfId, timer); } this.UpdateButtons(wfId, true); this._timers[wfId].Start(); } else { this.UpdateButtons(wfId, true); } }}
当的选择DataGridView
变化,为选定工作流程创建一个计时器,如果它不存在,那么这个计时器更新Enabled
按钮的选项和 textBoxInfo
文本框,如果所选择的工作流的状态通过时间的变化。
如果在Wexflow Manager中双击工作流,它将启动,如果暂停,则恢复如下:
private void dataGridViewWorkflows_CellDoubleClick(object sender, DataGridViewCellEventArgs e){ int wfId = this.GetSlectedWorkflowId(); if (wfId > -1) { Workflow workflow = _wexflowEngine.GetWorkflow(wfId); if (workflow != null && workflow.IsEnabled) { if (!workflow.IsRunning && !workflow.IsPaused) { buttonStart_Click(this, null); } else if(workflow.IsPaused) { buttonResume_Click(this, null); } } }}
而已。我希望你喜欢阅读这篇文章。如果你有任何想法,以改善Wexflow或如果你面对任何问题,或者如果你想贡献给这个项目,请让我知道在评论。
这里是Wexflow使用的库列表:
Tar
,Tgz
和Sql
任务。Wmi
和 ImagesTransformer
任务。Http
, Sync
, FilesRenamer
, FilesExist
和 Wait
任务。Ftp
任务(FTP / FTPS(显性/隐性)/ SFTP)。retryCount
和retryTimeout
设置选项Ftp
的任务。
联系客服