从开源的ESB项目，基本上都是java版本，c#的esb开源项目太少，这对搞c#来说是一大遗憾，所以本人很想对c#开源世界做一份小小的贡献，因些开始写一版真正开源的ESB企业开发框架。园子里有兴趣的园友也可以一起来做。
1,ESB架构图
2,ESB系统架构要求
Windows Server 2003 R2
Windows Server 2008
.Net 4.0 Framework
处理器要求: 最少1 GHz or faster
内存要求: 最少2 GB RAM
服务器需要打开的端口:
Redis服务端口: 6379~6384
JS, WP7, Java API 端口: 8078
router and gateway service 端口:8000, 8085, and 8084
websocket 端口:8081
3,ESB配置
负载配置说明(LoadBalance)
4,ESB功能说明
(1)实现集成权限认证提供安全管理
(2)支持数据交换方式有:
WCF(NetTcp),MSSQL,WP7,MSMQ,RabbitMQ,Redis,TCP,WebService,Http,Email 等
(3)可插拔的接口
(4)提供简单API访问ESB
(5)开发者可以通过Javascript API与ESB交互
(6)提供REST 服务与ESB进行交互
5,ESB测试结果
测试机器
Intel Core 2 Quad Q9000 @2.00GHZ 6GB RAM x64bit
压力测试结果: StressRequest - Result
Published Message to Bus in 0.2240128 Second(s)
Finished processing(publishing/receiving) 1000 in 0 second(s)
Published Message to Bus in 2.0591178 Second(s)
Finished processing(publishing/receiving) 10000 in 2.4851422 second(s)
Published Message to Bus in 22.8293057 Second(s)
Finished processing(publishing/receiving) 100000 in 23.4173394 second(s)
Published Message to Bus in 236.3805202 Second(s)
Finished processing(publishing/receiving) 1000000 in 237.2985727 second(s)
Published Message to Bus in 2440.2385737 Second(s)
Finished processing(publishing/receiving) 10000000 in 2440.4655866 second(s)
6,序列化方案
MessageShark(Open Source Binary Serializer to replace .Net Binary Formatter)
7,操作说明
1,首先配置ESB指向地址总线
(1),配置负载地址
//Configure router fail-over addresses in cases where multiple router services exists
ESB.AddRouterAddress("tcp://127.0.0.1:8001");
ESB.AddRouterAddress("tcp://127.0.0.1:8002");
(2),配置默认ESB路由结点：localhost:8000/ESBGateway
//Configure default ESB router endpoint to localhost:8000/ESBGateway
ESB.ConfigWithAddress("tcp://127.0.0.1:8000")
2,设置服务认证安全与权限
为了控制谁连接或者调用到ESB服务,你需要实现两个接口:IAuthenticationProvider和 IAuthorizationProvider
ESB.Authenticate("username", "password");
下面是实现了这两个接口的例子,这两个引用需要引用PServiceBus.Core.dll
Public class DefaultAuthenticationProvider:IAuthenticationProvider
{
Public bool Authenticate(string username,string password)
{
//添加你的验证用户的业务逻辑:例如连接数据库或者其它
}
}
public class DefaultAuthorizationProvider : IAuthorizationProvider
{
public bool CanEditSubscriber(string username, string subscriberName)
{
//Do something to verify that user can edit
specified subscriber information
}
}
3,注册一个新的话题(topic)
什么是Topic(话题)
一个主题包括名称，描述和参数。主题名称的方法.
Contract Topic :需要严格参数要求
Open Topic: 发布者可以发布任意参数
Dynamic Topic Creation (动态话题创建)
Topic.New("MyTopic")
.Description("My Topic Description")
.AddParameter("Param1", "Param1 Description")
.AddParameter("Param2", "Param2 Description")
.Register();
Class Based Topic Creation (静态话题创建)
[Description("My topic Description")]
public class MyTopic {
[Description("Param1 Description")]
public string Param1 { get; set; }
[Description("Param2 Description")]
public string Param2 {get; set;}
}
Topic.Register<MyTopic>();
4,发布一个消息给一个话题(Topic)
给一个话题发布消息有两个方式:你可以发布一个单个的消息，或者你可以发布一个消息集合,你要发布一个消息，你要定义这个消息你想发布到哪个话题,你可以通过选择是通过名称或者类型()来查找话题。
Retrieve Topic object by using the type name as the topic name
var topic = Topic.Select<MyTopic>()
//Retrieve Topic object by using a string based name
var topic = Topic.Select("MyTopic")
发布一个单个消息
There are fours ways to go about publishing a single message.
//Explicitly add the parameter for topic message
topic.SetParameter("Param1", "Value1")
.SetParameter("Param2", "Value2")
.Publish()
//Map an anonymous object or any instance of a class that matches the topic contract
//This option allow you to build up the parameter that would be published from one or more object as needed
topic.Message(new {Param1 = "Value1", Param2 = "Value2"})
.Publish()
//Map with instance of a class
topic.Message(new MyTopic(){ Param1 = "Value1", Param2 = "Value2" })
.Publish();
//Publish using anonymous object
topic.Publish(new {Param1 = "Value1", Param2 = "Value2"})
//Publish using instance of a class
topic.Publish(new MyTopic(){ Param1 = "Value1", Param2 = "Value2" })
//Publish message and infer the specific topic to publish to by using the class name
//This approach does not need the instance of topic object since it is inferring the topic name
Topic.PublishMessage(new MyTopic(){ Param1 = "Value1", Param2 = "Value2"})
给话题发布多个消息体
There are three ways to publish messages in batches
//Publish using a collection of anonymous objects
topic.PublishMany(new List<object>(){
new {Param1 = "Value1", Param2 = "Value2"},
new {Param1 = "Value1", Param2 = "Value2"}
} );
//Publish using a collection of instance of a class
topic.PublishMany(new List<MyTopic>(){
new MyTopic(){ Param1 = "Value1", Param2 = "Value2"},
new MyTopic(){ Param1 = "Value1", Param2 = "Value2"}
});
//Publish by infer the topic name from the type constrain of the collection
Topic.PublishMessages(new List<MyTopic>(){
new MyTopic(){ Param1 = "Value1", Param2 = "Value2"},
new MyTopic(){ Param1 = "Value1", Param2 = "Value2"}
});
6,获取所有的话题
如果你想知道目前ESB中有多少话题(topics),可以通过下面的方法来获取,ESB会反回一个集合，包括话题的名称,描述，参数。
ESB.GetTopics()
7,对一个话题进行消费(subscription)
要开始接收发布给话题的消息，你要配置参数(filters, transports,topic)
var topicName = "ChatTopic";
var subscriberName = "MySubscriber";
Subscriber.New(subscriberName).Durable(false)
.SubscribeTo(Topic.Select<ChatTopic>().NotEqual("UserName", _userName))
.AddTransport("Tcp", Transport.New<TcpTransport>(
transport => {
transport.Format = TransportFormat.Json;
transport.IPAddress = "127.0.0.1";
transport.Port = port;
}), topicName)
.Save()
以下解释一个每个方法的意思:
Create a new instance of subscriber with the name supplied in the New method
Subscriber.New(subscriberName)
如果消息发送失败，esb会把失败的消息进行保存，默认情况下，每个消息保存30天不过期
subscriber.SubscribeTo(Topic.Select<ChatTopic>().NotEqual("UserName", _userName))
subscriber.AddTransport("Tcp", Transport.New<TcpTransport>(
transport => {
transport.Format = TransportFormat.Json;
transport.IPAddress = "127.0.0.1";
transport.Port = port;
}), topicName)
subscriber.Save()
为一个消费者(subscriber)在传输协议上添加消息监听
subscriber.OnMessageReceived<ChatTopic>("Tcp", msg => HandleMessage(msg),
interval: TimeSpan.FromMilliseconds(1),
errorAction: h => {
h.Continue = true;
Console.WriteLine(h.Error.ToString());
});
方法解释说明:
8,把消息存储服务更新到数据库中
<add name="default" type="ServiceBus.MessageStores.DefaultMessageStore"
address="Server=.\SQLExpress;Database=servicebus;Trusted_Connection=yes;MultipleActiveResultSets=True"/>
创建数据库，执行下面的SQL语句
Src\ServiceBus.Data.Entities\Scripts\SentMessageStorage.sql