Âë ÊղشúÂë
public Publisher() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL);
connection = factory.createConnection();
try {
connection.start();
} catch (JMSException jmse) {
connection.close();
throw jmse;
}
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(null);
}

protected void setTopics(String[] stocks) throws JMSException {
destinations = new Destination[stocks.length];
for(int i = 0; i < stocks.length; i++) {
destinations[i] = session.createTopic("STOCKS." + stocks[i]);
}
}

protected void sendMessage(String[] stocks) throws JMSException {
for(int i = 0; i < stocks.length; i++) {
Message message = createStockMessage(stocks[i], session);
System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destinations[i]);
producer.send(destinations[i], message);
}
}

protected Message createStockMessage(String stock, Session session) throws JMSException {
MapMessage message = session.createMapMessage();
message.setString("stock", stock);
message.setDouble("price", 1.00);
message.setDouble("offer", 0.01);
message.setBoolean("up", true);

return message;
}

public static void main(String[] args) throws JMSException {
if(args.length < 1)
throw new IllegalArgumentException();

// Create publisher
Publisher publisher = new Publisher();

// Set topics
publisher.setTopics(args);

for(int i = 0; i < 10; i++) {
publisher.sendMessage(args);
System.out.println("Publisher '" + i + " price messages");
try {
Thread.sleep(1000);
} catch(InterruptedException e) {
e.printStackTrace();
}
}
// Close all resources
publisher.close();
}

public void close() throws JMSException {
if (connection != null) {
connection.close();
}
}

public Consumer() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL);
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
}

public static void main(String[] args) throws JMSException {
Consumer consumer = new Consumer();
for (String stock : args) {
Destination destination = consumer.getSession().createTopic("STOCKS." + stock);
MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);
messageConsumer.setMessageListener(new Listener());
}
}

public Session getSession() {
return session;
}

public class Listener implements MessageListener {

public void onMessage(Message message) {
try {
MapMessage map = (MapMessage)message;
String stock = map.getString("stock");
double price = map.getDouble("price");
double offer = map.getDouble("offer");
boolean up = map.getBoolean("up");
DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" );
System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down"));
} catch (Exception e) {
e.printStackTrace();
}
}

}

public Publisher() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL);
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producer = session.createProducer(null);
}

public void sendMessage() throws JMSException {
for(int i = 0; i < jobs.length; i++)
{
String job = jobs[i];
Destination destination = session.createQueue("JOBS." + job);
Message message = session.createObjectMessage(i);
System.out.println("Sending: id: " + ((ObjectMessage)message).getObject() + " on queue: " + destination);
producer.send(destination, message);
}
}

public static void main(String[] args) throws JMSException {
Publisher publisher = new Publisher();
for(int i = 0; i < 10; i++) {
publisher.sendMessage();
System.out.println("Published " + i + " job messages");
try {
Thread.sleep(1000);
} catch (InterruptedException x) {
e.printStackTrace();
}
}
publisher.close();
}

public Consumer() throws JMSException {
factory = new ActiveMQConnectionFactory(brokerURL);
connection = factory.createConnection();
connection.start();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
}

public static void main(String[] args) throws JMSException {
Consumer consumer = new Consumer();
for (String job : consumer.jobs) {
Destination destination = consumer.getSession().createQueue("JOBS." + job);
MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);
messageConsumer.setMessageListener(new Listener(job));
}
}

public Session getSession() {
return session;
}

import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

public class Listener implements MessageListener {

private String job;

public Listener(String job) {
this.job = job;
}

public void onMessage(Message message) {
try {
//do something here
System.out.println(job + " id:" + ((ObjectMessage)message).getObject());
} catch (Exception e) {
e.printStackTrace();
}
}

}

// client side
Destination tempDest = session.createTemporaryQueue();
MessageConsumer responseConsumer = session.createConsumer(tempDest);
...

// send a request..
message.setJMSReplyTo(tempDest)
message.setJMSCorrelationID(myCorrelationID);

producer.send(message);

public void onMessage(Message request) {

Message response = session.createMessage();
response.setJMSCorrelationID(request.getJMS

CorrelationID())

producer.send(request.getJMSReplyTo(), response)
}

public Client() {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection;
try {
connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(transacted, ackMode);
Destination adminQueue = session.createQueue(clientQueueName);

//Setup a message producer to send message

to the queue the server is consuming from
this.producer = session.createProducer

(adminQueue);
this.producer.setDeliveryMode(DeliveryMode.

NON_PERSISTENT);

//Create a temporary queue that this client

will listen for responses on then create a

consumer
//that consumes message from this temporary

queue...for a real application a client

should reuse
//the same temp queue for each message

to the server...one temp queue per client
Destination tempDest = session.create

TemporaryQueue();
MessageConsumer responseConsumer = session.createConsumer(tempDest);

//This class will handle the messages to

the temp queue as well
responseConsumer.setMessageListener(this);

//Now create the actual message you want to send
TextMessage txtMessage = session.create

TextMessage();
txtMessage.setText("MyProtocolMessage");

//Set the reply to field to the temp

queue you created above, this is the queue the server
//will respond to
txtMessage.setJMSReplyTo(tempDest);

//Set a correlation ID so when you get a

response you know which sent message the

response is for
//If there is never more than one

outstanding message to the server then the
//same correlation ID can be used for all

the messages...if there is more than one

outstanding
//message to the server you would presumably

want to associate the correlation ID with this
//message somehow...a Map works good
String correlationId = this.createRandomString();
txtMessage.setJMSCorrelationID(correlationId);
this.producer.send(txtMessage);
} catch (JMSException e) {
//Handle the exception appropriately
}
}

public void onMessage(Message message) {
String messageText = null;
try {
if (message instanceof TextMessage) {
TextMessage textMessage = (TextMessage) message;
messageText = textMessage.getText();
System.out.println("messageText = " + messageText);
}
} catch (JMSException e) {
//Handle the exception appropriately
}
}

public void onMessage(Message message) {
try {
TextMessage response = this.session.create

TextMessage();
if (message instanceof TextMessage) {
TextMessage txtMsg = (TextMessage) message;
String messageText = txtMsg.getText();
response.setText(this.messageProtocol

.handleProtocolMessage(messageText));
}

//Set the correlation ID from the received

message to be the correlation id of the

response message
//this lets the client identify which message

this is a response to if it has more than
//one outstanding message to the server
response.setJMSCorrelationID(message.get

JMSCorrelationID());

//Send the response to the Destination

specified by the JMSReplyTo field of

the received message,
//this is presumably a temporary queue

created by the client
this.replyProducer.send(message.

getJMSReplyTo(), response);
} catch (JMSException e) {
//Handle the exception appropriately
}
}

public Server() {
try {
//This message broker is embedded
BrokerService broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(false);
broker.addConnector(messageBrokerUrl);
broker.start();
} catch (Exception e) {
//Handle the exception appropriately
}

//Delegating the handling of messages to

another class, instantiate it before

setting up JMS so it
//is ready to handle messages
this.messageProtocol = new MessageProtocol();
this.setupMessageQueueConsumer();
}

private void setupMessageQueueConsumer() {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(messageBrokerUrl);
Connection connection;
try {
connection = connectionFactory.createConnection();
connection.start();
this.session = connection.createSession

(this.transacted, ackMode);
Destination adminQueue = this.session.createQueue(messageQueueName);

//Setup a message producer to respond to

messages from clients, we will get the destination
//to send to from the JMSReplyTo header

field from a Message
this.replyProducer = this.session.create

Producer(null);
this.replyProducer.setDeliveryMode(Deliv

eryMode.NON_PERSISTENT);

//Set up a consumer to consume messages

off of the admin queue
MessageConsumer consumer = this.session.createConsumer(adminQueue);
consumer.setMessageListener(this);
} catch (JMSException e) {
//Handle the exception appropriately
}
}