使用 “离线事件” 处理 “长事务”

使用 “离线事件” 处理 “长事务”

背景

事件有两种使用方式：一、作为传统的监听者模式以达到程序结构的解耦；二、作为消息机制以达到时间和空间上的解耦，如发送到远程服务器、持久化到队列等待。今天介绍如何使用“离线事件”处理“长事务”，这就需要把事件当做消息对待。

我理解的长事务是“执行时间长的任务，具体多少没有标准”，如果希望在一个数据库事务中完成这些长事务是不现实的，之前我的做法是换成存储过程以降低事务的执行时间，以后我会采用“离线事件”。

离线事件：事件的一部分是同步执行，另外一部分会被异步的离线的在另外一台机器执行。

简单示例

下载地址：OfflineEventStudy。

项目结构

Common：类库，包含了事件和事件监听者（同步事件监听者和离线事件监听者）。

Console：离线事件执行程序，定时从队列取事件并执行离线事件监听者。

Web：事件生成程序，发布事件并执行同步事件监听者。

Common中的代码

1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Threading.Tasks;
6
7 using Happy.Event;
8 using Happy.Event.Offline;
9
10 namespace OfflineEventStudy.Common
11 {
12     [Persistable(1)]
13     public sealed class TestEvent : IEvent
14     {
15         public const string FilePath = @"E:\log.txt";
16
17         public string Info { get; set; }
18     }
19 }

注意：Persistable特性会导致此事件支持离线订阅。

1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Threading.Tasks;
6 using System.IO;
7 using System.Threading;
8
9 using Happy.Event;
10 using Happy.Event.Offline;
11
12 namespace OfflineEventStudy.Common
13 {
14     public sealed class TestEventSubscriber : ISyncEventSubscriber<TestEvent>, IOfflineEventSubscriber<TestEvent>
15     {
16         void ISyncEventSubscriber<TestEvent>.Handle(TestEvent @event)
17         {
18             File.WriteAllText(TestEvent.FilePath, @event.Info + "：任务正在执行中！");
19         }
20
21         void IOfflineEventSubscriber<TestEvent>.Handle(TestEvent @event)
22         {
23             Thread.Sleep(5000);
24
25             File.WriteAllText(TestEvent.FilePath, @event.Info + "：任务完成！");
26         }
27     }
28 }

注意：同步接口会在Web中执行，离线接口会在Console中执行。

Console中的代码

1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Text;
5 using System.Threading.Tasks;
6 using System.Threading;
7
8 using Happy.Event;
9 using Happy.Event.Offline;
10 using Happy.Bootstrap;
11 using Happy.Infrastructure.Unity;
12
13 namespace OfflineEventStudy.Console
14 {
15     class Program
16     {
17         static void Main(string[] args)
18         {
19             /****************************************启动过程配置****************************************/
20
21             BootstrapService
22              .Current
23              .IntegrateWithUnity() //使用Unity作为Ioc容器。
24              .UseRegisterServiceByConventionPlug() //使用按照约定注册服务插件，会自动帮你执行注册。
25                 .UseEventSubscriberRegister() //注册所有的EventListener。
26                 .Done() //完成配置。
27              .Start(); //启动。
28
29             /****************************************启动过程配置****************************************/
30
31             var processor = new OfflineEventProcessor(OfflineEventQueueFactory.CreateMSMQOfflineEventQueue("OfflineEventStudy"));
32
33             ThreadPool.QueueUserWorkItem((state) =>
34             {
35                 processor.Start();
36             }, null);
37
38             System.Console.WriteLine("正在监听任务");
39             System.Console.Read();
40         }
41     }
42 }

注意：这里会不断的从队列取事件并执行。

Web中的代码

1 using System;
2 using System.Collections.Generic;
3 using System.Linq;
4 using System.Web;
5 using System.Web.UI;
6 using System.Web.UI.WebControls;
7 using System.IO;
8
9 using Happy.Event;
10 using OfflineEventStudy.Common;
11
12 namespace OfflineEventStudy.Web
13 {
14     public partial class Default : System.Web.UI.Page
15     {
16         protected void Page_Load(object sender, EventArgs e)
17         {
18             if (this.IsPostBack)
19             {
20                 return;
21             }
22
23             this.Display();
24         }
25
26         protected void Button1_Click(object sender, EventArgs e)
27         {
28             EventPublisher.Current.Publish(new TestEvent { Info = "测试事件" });
29
30             this.Display();
31         }
32
33         protected void Button2_Click(object sender, EventArgs e)
34         {
35             this.Display();
36         }
37
38         private void Display()
39         {
40             if (!File.Exists(TestEvent.FilePath))
41             {
42                 return;
43             }
44
45             this.Label1.Text = File.ReadAllText(TestEvent.FilePath);
46         }
47     }
48 }

运行效果

备注

因为事件的一部分是离线执行的，所以不能保证实时一致性，关于最终一致性的问题这里不好多说，有兴趣的朋友可以找http://www.cnblogs.com/netfocus/聊聊，netfocus最新力作enode完全放弃实时一致性，换来的是高并发。