PubSub (Publish-Subscribe) kann man sehr gut dort einsetzen, wo eine beliebige Anzahl von Empfängern von Nachrichten existiert. Ab und zu wird PubSub auch als Observer-Pattern bezeichnet. Durch PubSub sollen alle die Interessiert sind, die Informationen der Veröffentlichung (Publish) erhalten. Der Vorgang der Veröffentlichung ist als asynchroner Mechanismus konzipiert, es kann allerdings auch synchron benutzt werden (wird aber nicht empfohlen). Aufgrund der asynchronen Ausführung ist der Einsatz einer Message Queue in vielen Anwendungsszenarien empfehlenswert. Eine Message Queue bringen viele Windows Server Systeme bereits mit (Msmq), so dass meistens keine weiteren Kosten anfallen, alternativ kann auch eine Datenbank verwendet werden.
Binding, die Duplex-Verbindungen unterstützen, sind:
- WSDualHttpBinding
- NetTcpBinding
- NetNamedPipeBinding
- NetPeerTcpBinding
- NetTcpContextBinding
Auf der MSDN-Seite ist eine sehr gute Übersicht über die gebräuchlichsten Bindings und deren Fähigkeiten.
Mein Beispiel, hier der Download, ist nicht sehr sinnvoll aber es zeigt die grundsätzlichen Funktionen und einige Problemecken.
Ich habe ein sehr recht einfaches Klassenmodell, es werden simple mathematische Operationen in den Services durchgeführt. Es gibt einen Service zum Addieren und einen zum Multiplizieren. Es gibt 4 Service-Interfaces IMathAddService, IMathMultiplyService, ISubscriptionService und IPublishService. Die ersten beiden Services klammere ich aktuell etwas aus, entscheidend ist der ISubscriptionService bei dem sich alle Aufrufer (Subscriber) registrieren, um über Zustandsänderung/Veröffentlichungen informiert zu werden. In meinem Fall implementieren sowohl die Subscriber als auch der SubscriptionService das IPublishService-Interface. Es wäre nicht notwendig, dass der SubscriptionService das Interface implementiert, es macht es in meinem Fall aber einfacher. Durch die Publish-Methode im SubscriptionService werden die Veröffentlichungen angestoßen. Der SubscriptionService führt Veröffentlichungen allerdings nicht sofort aus, sonder stellt diese in eine Queue, da ich immer schön nacheinander eine Abarbeitung durchführen möchte. Die Queue ist im SubscriberManager mit implementiert, typischerweise würde sich hier ein MessageQueuing-System anbieten.
1: //setup endpoints
2: EndpointAddress addressAdd =
3: new EndpointAddress("net.tcp://localhost:5554/pubsub/add");
4: EndpointAddress addressMultiply =
5: new EndpointAddress("net.tcp://localhost:5554/pubsub/mutliply");
6: EndpointAddress addressPubSub =
7: new EndpointAddress("net.tcp://localhost:5554/pubsub/register");
8: //binding with duplex support is required! (WSDualHttpBinding, NetTcpBinding, ..)
9: NetTcpBinding binding = new NetTcpBinding();
10:
11: //create the channels to register the subscribers (link to subscriber interface)
12: DuplexChannelFactory<ISubscriptionService> duplexAddService, duplexMultiplyService;
13: duplexAddService = new DuplexChannelFactory<ISubscriptionService>(
14: new InstanceContext(new MathAddService()), binding, addressPubSub);
15: duplexMultiplyService = new DuplexChannelFactory<ISubscriptionService>(
16: new InstanceContext(new MathMultiplyService()), binding, addressPubSub);
17:
18: //create all host and start them
19: ServiceHost hostAdd = new ServiceHost(typeof(MathAddService));
20: ServiceHost hostMultiply = new ServiceHost(typeof(MathMultiplyService));
21: ServiceHost hostSubscribe = new ServiceHost(typeof(MemoryPubSubService));
22: hostAdd.AddServiceEndpoint(typeof(IMathAddService), binding, addressAdd.Uri);
23: hostMultiply.AddServiceEndpoint(typeof(IMathMultiplyService)
24: , binding, addressMultiply.Uri);
25: hostSubscribe.AddServiceEndpoint(typeof(ISubscriptionService)
26: , binding, addressPubSub.Uri);
27: hostAdd.Open();
28: hostMultiply.Open();
29: hostSubscribe.Open();
30:
31: //register!
32: ISubscriptionService dupAdd = duplexAddService.CreateChannel();
33: dupAdd.Subscribe(PublishTopics.All);
34: duplexMultiplyService.CreateChannel().Subscribe(PublishTopics.All);
35: //run the watcher (async processor)
36: using (PublishMonitor monitor = new PublishMonitor())
37: {
38:
39: ChannelFactory<IMathAddService> addFactory =
40: new ChannelFactory<IMathAddService>(binding, addressAdd);
41: ChannelFactory<IMathMultiplyService> multiplyFactory =
42: new ChannelFactory<IMathMultiplyService>(
43: binding, addressMultiply);
44: IMathAddService add = addFactory.CreateChannel();
45: IMathMultiplyService multiply = multiplyFactory.CreateChannel();
46:
47: Console.WriteLine("Adding 1+2=" + add.Add(1, 2));
48: Console.WriteLine("LastResult of add={0}, of multiply={1}", add.GetLastResult(),
49: multiply.GetLastResult());
50: Console.WriteLine("Calling publish!");
51: //this call should be normally done by the service itself
52: dupAdd.Publish(new PublishContext()
53: {
54: Topic = PublishTopics.Add, SomeDetail = "Some one to reset!"
55: });
56: dupAdd.Publish(new PublishContext()
57: {
58: Topic = PublishTopics.Multiply, SomeDetail = "Some want's to reset again!"
59: });
60: //show some concurrency probs, that can occure
61: Console.WriteLine("LastResult of add={0}, of multiply={1}", add.GetLastResult(),
62: multiply.GetLastResult());
63: Console.WriteLine("Adding 4*2=" + multiply.Multiply(4, 2));
64:
65: Thread.Sleep(2000);
66: Console.WriteLine("LastResult of add={0}, of multiply={1}", add.GetLastResult(),
67: multiply.GetLastResult());
68: Console.WriteLine("Finished sample, Press [Enter]");
69: Console.ReadLine();
70: Console.WriteLine("closing proxies and hosts");
71: addFactory.Close();
72: multiplyFactory.Close();
73: }
74: //close all and cleanup
75: duplexAddService.Close();
76: duplexMultiplyService.Close();
77: hostSubscribe.Close();
78: hostAdd.Close();
79: hostMultiply.Close();
Wird das Ganze ausgeführt, so erhalte ich zum Beispiel auf meinem Dual Core das folgende Ergebnis:
Perfekt ist die Implementierung meines Beispiels nicht, allerdings soll es auch nur einige Möglichkeiten und Probleme aufzeigen.
Hier noch einige interessante Links zu dem Thema:
- P&P Publish/Subscribe http://msdn.microsoft.com/en-us/library/ms978603.aspx
- EAI Patterns Beschreibung von Publish/Subscribe http://www.eaipatterns.com/PublishSubscribeChannel.html
- Ein Codeproject Beispiel für http://www.codeproject.com/KB/IP/PubSubUsingWCF.aspx
- Nochmals der Beispiel Code Download http://www.zieschang-jan.de/documents/PublishSubscribeTest.zip
