支持异步同步的分布式CommandBus MSMQ实现 - 支持Session传递、多实例处理
2013-09-05 00:38
579 查看
先上一张本文所描述的适用场景图
分布式场景,共3台server:
前端Server
Order App Server
Warehouse App Server
功能:
前端Server可以不停的发送Command到CommandBus,然后由CommandBus分配不同的Command到各自的app server去处理。
前端Server可以只发送Command而不必等待Response
前端Server可以同步等待Response返回
MSMQ消息超过3.5M会自动转为网络共享方式传输消息
对于同一Command的处理,可以通过增加App Server的方式来提高并发处理速度(比如:可以开2个app server instance来同时处理ACommand的处理)
在App server中能读取前端Server的Session value(写Session还不支持)
本文目标是用msmq实现分布式CommandBus的应用(已经更新到A2D Framework中了)。
前端Server发送Command的方式(异步):
同步方式:
配置文件:
Command的编写方式:
后端App Server的编写
CommandHandlerHost1: Console程序,相当于App Server 1,会处理部分Command
CommandHandlerHost2: Console程序,相当于App Server 2,会处理部分Command
CommandHandlers: 所有的Command处理函数都会在这个项目中实现
下面是目前的性能测试:
下载代码
分布式场景,共3台server:
前端Server
Order App Server
Warehouse App Server
功能:
前端Server可以不停的发送Command到CommandBus,然后由CommandBus分配不同的Command到各自的app server去处理。
前端Server可以只发送Command而不必等待Response
前端Server可以同步等待Response返回
MSMQ消息超过3.5M会自动转为网络共享方式传输消息
对于同一Command的处理,可以通过增加App Server的方式来提高并发处理速度(比如:可以开2个app server instance来同时处理ACommand的处理)
在App server中能读取前端Server的Session value(写Session还不支持)
本文目标是用msmq实现分布式CommandBus的应用(已经更新到A2D Framework中了)。
前端Server发送Command的方式(异步):
ACommand cmd = new ACommand() { Tag = "aaa" }; CommandBusDistributer<ACommand, ACommandResult> cmdDistributer = new CommandBusDistributer<ACommand, ACommandResult>(); cmdDistributer.ResultCatached += new CommandResultCatchedDelegate<ACommandResult>(cmdDistributer_ResultCatached); cmdDistributer.SendRequest(cmd);
同步方式:
ACommand cmd = new ACommand() { Tag = "aaa" }; CommandBusDistributer<ACommand, ACommandResult> cmdDistributer = new CommandBusDistributer<ACommand, ACommandResult>(); cmdDistributer.SendRequest(cmd); ACommandResult result=cmdDistributer.WaitResponse();
配置文件:
<?xml version="1.0" encoding="utf-8" ?> <CommandBusSetting> <AutoCreateIfNotExists>true</AutoCreateIfNotExists> <WebSessionSupport>true</WebSessionSupport> <CommandQueue>PC-20130606HCVP\private$\Commands_{0}</CommandQueue> <ResponseQueue>PC-20130606HCVP\private$\CommandResponses</ResponseQueue> <NetworkLocation>\\PC-20130606HCVP\network</NetworkLocation> </CommandBusSetting>
Command的编写方式:
[QueueName("ACommand")]//这个可选,没有QueueName时,默认对应的msmq队列名为类名,此处为ACommand public class ACommand : BaseCommand //需要继承自BaseCommand { public string Tag { get; set; }//自定义的属性 }
后端App Server的编写
CommandHandlerHost1: Console程序,相当于App Server 1,会处理部分Command
static void Main(string[] args) { Thread.Sleep(2000); CommandHandlerListener listener = new CommandHandlerListener(); listener.AddHandler(new TestCommandHandlers()); listener.AddHandler(new Test2CommandHandlers()); listener.Start(); Console.ReadKey(); }
CommandHandlerHost2: Console程序,相当于App Server 2,会处理部分Command
static void Main(string[] args) { Thread.Sleep(2000); CommandHandlerListener listener = new CommandHandlerListener(); listener.AddHandler(new Test3CommandHandlers()); listener.Start(); Console.ReadKey(); }
CommandHandlers: 所有的Command处理函数都会在这个项目中实现
public class TestCommandHandlers : ICommandHandlers, ICommandHandler<ACommand, ACommandResult>, ICommandHandler<BCommand, BCommandResult> { public ACommandResult Handler(ACommand cmd, ISessionContext session) { Console.WriteLine("From [public ACommandResult Handler(ACommand cmd)]: " + cmd.Tag); ACommandResult result = new ACommandResult(); result.Result = "result from ACommand"; return result; } public BCommandResult Handler(BCommand cmd, ISessionContext session) { Console.WriteLine("From [public BCommandResult Handler(BCommand cmd)]: " + cmd.Tag); if (session != null) { Console.WriteLine("session not null"); object o = session.GetWebSession("key1"); if (o != null) { Console.WriteLine("From Session['key1']: " + Convert.ToString(o)); //session.Set("key1", "changed from command handler: " + DateTime.Now); } } BCommandResult result = new BCommandResult(); result.Result = "result from BCommand"; return result; } }
下面是目前的性能测试:
下载代码
相关文章推荐
- session失效时间设置、过期处理、ajax同步异步请求处理
- 应用层和驱动层的同步与异步的处理逻辑及底层实现
- ajax 实现 GET POST 异步 同步 检验mysql连接情况简单实例
- VC实现服务端和多客户端同步、超时处理实例
- 应用层和驱动层的同步与异步的处理逻辑及底层实现
- 应用层和驱动层的同步与异步的处理逻辑及底层实现
- ajax前置处理实现异步请求session过期时跳转登录页面
- 使用Promise和async-await实现的一个异步遍历+同步执行任务的实例
- 支持异步同步的分布式CommandBus MSMQ实现
- nginx+tomcat+redis集群实现负载均衡和session同步的步骤和问题处理方法
- 批处理实现文件夹同步
- libcurl基础教程及封装(支持同步异步请求,支持多线程下载,支持https)
- php+js实现异步图片上传实例分享
- NSURLConnection和NSMutableURLRequest 实现同步、异步请求
- 微信小程序中使用Promise进行异步流程处理的实例详解
- Android异步处理四:AsyncTask的实现原理
- ajax异步请求,session超时处理
- C#一般处理程序设置和读取session(session报错未将对象引用设置到对象的实例解决)
- HTTP 通信, 三种方式XML 解析,并通过 Hander 实现异步消息处理
- 使用querystring,session变量以及server.Transfer方法来实现页面间的值传递