Skip to content

Commit 0844618

Browse files
EhspressoEhspresso
committed
iluwatar#2898 Initial commit
1 parent a344f1d commit 0844618

File tree

9 files changed

+734
-0
lines changed

9 files changed

+734
-0
lines changed

‎publisher-subscriber/README.md

+358
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,358 @@
1+
---
2+
title: Publish-Subscribe pattern
3+
category: Behavioral
4+
language: en
5+
tag:
6+
- Distributed Systems
7+
- Messaging
8+
- JMS
9+
---
10+
11+
## Also known as
12+
13+
Pub-Sub Pattern
14+
15+
## Intent
16+
17+
To provide an asynchronous manner of handling messages between a message producer and a message consumer, without coupling the producer and consumer.
18+
19+
## Explanation
20+
21+
Real-world example
22+
23+
> In online multipler games, players need real-time updates on player actions and game events. When updates needs to be broadcasted to multiple players the players act as subscribers to receive these updates. Depending on player level, account restrictions and so forth, a given player receives updates that another player may not receive and vice versa.
24+
25+
In plain words
26+
27+
> Publishers send messages that pertain to a certain topic and all subscribers to that topic will receive the message.
28+
29+
Wikipedia says
30+
31+
> Publish–subscribe is a messaging pattern where publishers categorize messages into classes that are received by subscribers. This is contrasted to the typical messaging pattern model where publishers send messages directly to subscribers.
32+
Similarly, subscribers express interest in one or more classes and only receive messages that are of interest, without knowledge of which publishers, if any, there are.
33+
Publish–subscribe is a sibling of the message queue paradigm, and is typically one part of a larger message-oriented middleware system
34+
35+
**Programmatic Example**
36+
37+
Consider an application in which a mortgage lender publishes interest rate updates to borrowers. When received, borrowers can decide whether they should refinance based on the updated rate.
38+
39+
First, we have the Lender class which acts as a publisher. The user can enter interest rates in which they will be published to all subscribers.
40+
41+
output:
42+
43+
Press enter to quit application
44+
Enter: Rate
45+
46+
e.g. 6.8
47+
48+
```java
49+
public class TLender {
50+
51+
private TopicConnection tConnection;
52+
private TopicSession tSession;
53+
private Topic topic;
54+
private TopicPublisher publisher;
55+
56+
public TLender(String topicCFName, String topicName) {
57+
58+
try {
59+
//create context and retrieve objects from directory
60+
Context context = new InitialContext();
61+
TopicConnectionFactory topicCF = (TopicConnectionFactory) context.lookup(topicCFName);
62+
tConnection = topicCF.createTopicConnection();
63+
64+
//create connection
65+
tSession = tConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
66+
67+
//Retrieve request and response queues
68+
topic = (Topic) context.lookup(topicName);
69+
70+
//Create publisher
71+
publisher = tSession.createPublisher(topic);
72+
73+
tConnection.start();
74+
} catch(NamingException e) {
75+
e.printStackTrace();
76+
LOGGER.error("An error has occurred!", e);
77+
System.exit(1);
78+
} catch(JMSException e) {
79+
e.printStackTrace();
80+
LOGGER.error("An error has occurred!", e);
81+
System.exit(1);
82+
}
83+
}
84+
85+
private void publishRate(double newRate) {
86+
87+
try {
88+
//create JMS message
89+
BytesMessage message = tSession.createBytesMessage();
90+
message.writeDouble(newRate);
91+
92+
//publish message
93+
publisher.publish(message);
94+
} catch(JMSException e) {
95+
e.printStackTrace();
96+
LOGGER.error("An error has occurred!", e);
97+
System.exit(1);
98+
}
99+
}
100+
101+
private void exit() {
102+
try {
103+
tConnection.close();
104+
} catch(JMSException e) {
105+
e.printStackTrace();
106+
LOGGER.error("An error has occurred!", e);
107+
System.exit(1);
108+
}
109+
System.exit(0);
110+
}
111+
112+
public static void main(String[] args) {
113+
114+
String topicCFName = null;
115+
String topicName = null;
116+
117+
if(args.length == 2) {
118+
topicCFName = args[0];
119+
topicName = args[1];
120+
} else {
121+
System.out.println("Invalid arguments. Should be: ");
122+
System.out.println("java TLender [factory] [topic]");
123+
System.exit(1);
124+
}
125+
126+
try {
127+
// Create and start activeMQ broker. Broker decouples publishers and subscribers.
128+
//Additionally brokers manage threads and asynchronous sending and receiving of messages.
129+
BrokerService broker = new BrokerService();
130+
broker.addConnector("tcp://localhost:61616");
131+
broker.start();
132+
133+
} catch(Exception e) {
134+
e.printStackTrace();
135+
LOGGER.error("An error has occurred!", e);
136+
}
137+
138+
TLender tLender = new TLender(topicCFName, topicName);
139+
140+
System.out.println ("TLender Application Started");
141+
System.out.println ("Press enter to quit application");
142+
System.out.println ("Enter: Rate");
143+
System.out.println("\ne.g. 6.8");
144+
145+
try {
146+
BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
147+
//Continuously read input and send as message to subscribers
148+
while(true) {
149+
System.out.print("> ");
150+
String line = reader.readLine();
151+
//Exit if user pressed enter or line is blank
152+
if (line == null || line.trim().length() == 0) {
153+
System.out.println("Exiting...");
154+
tLender.exit();
155+
}
156+
else { //publish the entered rate
157+
double newRate = Double.parseDouble(line);
158+
tLender.publishRate(newRate);
159+
}
160+
}
161+
} catch(IOException e) {
162+
e.printStackTrace();
163+
LOGGER.error("An error has occurred!", e);
164+
}
165+
}
166+
167+
public TopicConnection gettConnection() {
168+
return tConnection;
169+
}
170+
171+
public TopicSession gettSession() {
172+
return tSession;
173+
}
174+
175+
public Topic getTopic() {
176+
return topic;
177+
}
178+
}
179+
```
180+
181+
The Borrower class acts as a subscriber to a given topic, in this case mortgage interest rates. Evertime a new rate is published the subscriber receives the message.
182+
183+
output:
184+
185+
Initial rate is 6.0
186+
Waiting for new rates...
187+
Press enter to quit application
188+
189+
Running the class:
190+
191+
The class must be run after the TLender class is running since TLender spins up the activeMQ broker.
192+
193+
In order to see the messages being sent to multiple subscribers multiple instance of the TBorrower class need to be run. Either run multiple instances in an IDE or execute the following command in a command line from the root folder after generating the target folder:
194+
195+
196+
mvn exec:java -Dexec.mainClass=com.iluwatar.publishersubscriber.TBorrower -Dexec.args="TopicCF RateTopic 6"
197+
198+
```java
199+
public class TBorrower implements MessageListener {
200+
201+
private TopicConnection tConnection;
202+
private TopicSession tSession;
203+
private Topic topic;
204+
private double currentRate;
205+
206+
public TBorrower(String topicCFName, String topicName, double initialRate) {
207+
208+
currentRate = initialRate;
209+
210+
try {
211+
//create context and retrieve objects from directory
212+
Context context = new InitialContext();
213+
TopicConnectionFactory topicCF = (TopicConnectionFactory) context.lookup(topicCFName);
214+
tConnection = topicCF.createTopicConnection();
215+
216+
//create connection
217+
tSession = tConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
218+
219+
//Retrieve request and response queues
220+
topic = (Topic) context.lookup(topicName);
221+
222+
//Create subscriber and message listener
223+
TopicSubscriber subscriber = tSession.createSubscriber(topic);
224+
//Adds event listener to subscriber and uses onMessage as a callback
225+
subscriber.setMessageListener(this);
226+
227+
tConnection.start();
228+
System.out.println("Initial rate is " + currentRate + " \nWaiting for new rates...");
229+
} catch(NamingException e) {
230+
e.printStackTrace();
231+
LOGGER.error("An error has occurred!", e);
232+
System.exit(1);
233+
} catch(JMSException e) {
234+
e.printStackTrace();
235+
LOGGER.error("An error has occurred!", e);
236+
System.exit(1);
237+
}
238+
}
239+
240+
//This method is called asynchronously by the activeMQ broker
241+
public void onMessage(Message message) {
242+
243+
try {
244+
BytesMessage bMessage = (BytesMessage) message;
245+
double newRate = ((BytesMessage) message).readDouble();
246+
247+
if (currentRate - newRate >= 1)
248+
System.out.println("New Rate is " + newRate + " - Consider refinancing");
249+
else
250+
System.out.println("New Rate is " + newRate + " - Consider keeping current rate");
251+
} catch(JMSException e) {
252+
e.printStackTrace();
253+
LOGGER.error("An error occurred!", e);
254+
System.exit(1);
255+
}
256+
System.out.println("Waiting for new rates...");
257+
}
258+
259+
private void exit() {
260+
try {
261+
tConnection.close();
262+
} catch(JMSException e) {
263+
e.printStackTrace();
264+
LOGGER.error("An error has occurred!", e);
265+
System.exit(1);
266+
}
267+
System.exit(0);
268+
}
269+
270+
public static void main(String[] args) {
271+
272+
String topicCF = null;
273+
String topicName = null;
274+
int rate = 0;
275+
if (args.length == 3) {
276+
topicCF = args[0];
277+
topicName = args[1];
278+
rate = Integer.parseInt(args[2]);
279+
} else {
280+
System.out.println("Invalid arguments. Should be: ");
281+
System.out.println("java TBorrower [factory] [topic] [rate]");
282+
System.exit(0);
283+
}
284+
285+
TBorrower tBorrower = new TBorrower(topicCF, topicName, rate);
286+
287+
try {
288+
// Run until enter is pressed
289+
BufferedReader reader = new BufferedReader
290+
(new InputStreamReader(System.in));
291+
System.out.println ("TBorrower application started");
292+
System.out.println ("Press enter to quit application");
293+
reader.readLine();
294+
tBorrower.exit();
295+
} catch (IOException ioe) {
296+
ioe.printStackTrace();
297+
}
298+
}
299+
300+
public TopicConnection gettConnection() {
301+
return tConnection;
302+
}
303+
304+
public TopicSession gettSession() {
305+
return tSession;
306+
}
307+
308+
public Topic getTopic() {
309+
return topic;
310+
}
311+
312+
public double getCurrentRate() {
313+
return currentRate;
314+
}
315+
}
316+
```
317+
318+
## Class diagram
319+
320+
## Applicability
321+
322+
Use the Publish-Subscribe pattern when
323+
324+
* An application needs to broadcast information to a significant number of consumers.
325+
* An application needs to communicate with one or more independently developed applications or services, which may use different platforms, programming languages, and communication protocols.
326+
* An application needs to communicate information to multiple consumers, which may have different availability requirements or uptime schedules than the sender.
327+
328+
## Tutorials
329+
330+
* [Enterprise Integration Patterns](https://www.enterpriseintegrationpatterns.com/patterns/messaging/ObserverJmsExample.html)
331+
332+
## Consequences
333+
334+
Pros
335+
336+
* Pub-Sub activity is asynchronous (a.k.a, “fire and forget”). Hence, there is little risk of performance degradation due to a process getting caught in a long-running data exchange interaction.
337+
* Pub-sub promotes loose coupling between publishers and subscribers. Publishers and subscribers don't need to know each other's details, allowing for greater flexibility in system design and easier component replacement or updates.
338+
* Subscribers can dynamically subscribe and unsubscribe to topics based on their interests, allowing for dynamic adaptation to changing requirements or system conditions.
339+
* Pub-sub is well-suited for building event-driven architectures, where components react to events by subscribing to relevant topics. This enables real-time processing, event propagation, and system-wide coordination.
340+
341+
Cons
342+
343+
* Messaging systems typically don't guarantee strict message ordering across subscribers. While messages within a single topic are usually delivered in order, there's no guarantee of global message ordering across topics or subscribers.
344+
* Messaging introduces some latency compared to direct point-to-point communication, as messages need to be routed through the pub-sub system to reach subscribers. While typically minimal, this latency may be a consideration for latency-sensitive applications.
345+
* messages may be lost if subscribers are not actively consuming messages or if there are network or system failures.
346+
347+
## Real-world examples
348+
349+
* Market Data Feeds
350+
* Patient Monitoring in Healthcare
351+
* Supply Chain Visibility in Logistics
352+
* Communication between components in a distributed computer system
353+
354+
## Credits
355+
356+
* [Java Message Service, 2nd Edition](Author(s): Mark Richards, Richard Monson-Haefel, David A Chappell. Publisher(s): O'Reilly Media, Inc.)
357+
* [Red Hat](https://www.redhat.com/architect/pub-sub-pros-and-cons)
358+
* [Microsoft](https://learn.microsoft.com/en-us/azure/architecture/patterns/publisher-subscriber)

‎publisher-subscriber/pom.xml

+16
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,22 @@
5353
<artifactId>activemq-core</artifactId>
5454
<version>5.7.0</version>
5555
</dependency>
56+
<dependency>
57+
<groupId>org.apache.xbean</groupId>
58+
<artifactId>xbean-spring</artifactId>
59+
<version>4.24</version>
60+
</dependency>
61+
<dependency>
62+
<groupId>log4j</groupId>
63+
<artifactId>log4j</artifactId>
64+
<version>1.2.17</version>
65+
</dependency>
66+
<dependency>
67+
<groupId>org.apache.activemq.tooling</groupId>
68+
<artifactId>activemq-junit</artifactId>
69+
<version>6.1.1</version>
70+
<scope>test</scope>
71+
</dependency>
5672
</dependencies>
5773

5874
</project>

‎publisher-subscriber/src/main/java/com/iluwatar/publishersubscriber

Whitespace-only changes.

0 commit comments

Comments
 (0)