1:/*
2: * Small class to get started with ActiveMQ. Inspired by the examples contained in the
3: * ActiveMQ download. Gisbert Amm, 2008-10-28
4: */
5:package de.gisbertamm.activemqstart;
6:
7:import javax.jms.JMSException;
8:import org.apache.activemq.ActiveMQConnectionFactory;
9:import org.apache.activemq.broker.BrokerService;
10:import javax.jms.Connection;
11:import javax.jms.DeliveryMode;
12:import javax.jms.Destination;
13:import javax.jms.ExceptionListener;
14:import javax.jms.Message;
15:import javax.jms.MessageConsumer;
16:import javax.jms.MessageListener;
17:import javax.jms.MessageProducer;
18:import javax.jms.Session;
19:import javax.jms.TextMessage;
20:import org.apache.activemq.ActiveMQConnection;
21:
22:/**
23: *
24: * @author gisbert.amm
25: */
26:public class MyFirstActiveMqQueue {
27:
28: private static ActiveMQConnectionFactory connectionFactory;
29: private static Connection connection;
30: private static Session session;
31: private static Destination destination;
32: private static boolean transacted = false;
33:
34: public static void main(String[] args) throws Exception {
35: BrokerService broker = new BrokerService();
36: broker.setUseJmx(true);
37: broker.addConnector("tcp://localhost:61616");
38: broker.start();
39:
40: setUp();
41: createProducerAndSendAMessage();
42: System.out.println("Simulating a huge network delay :)");
43: Thread.sleep(4000);
44: createConsumerAndReceiveAMessage();
45:
46: //TODO: Find out how to get rid of the exceptions thrown when stopping the broker
47: broker.stop();
48: }
49:
50: private static void setUp() throws JMSException {
51: connectionFactory = new ActiveMQConnectionFactory(
52: ActiveMQConnection.DEFAULT_USER,
53: ActiveMQConnection.DEFAULT_PASSWORD,
54: ActiveMQConnection.DEFAULT_BROKER_URL);
55: connection = connectionFactory.createConnection();
56: connection.start();
57: session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
58: destination = session.createQueue("mmy first active mq queue");
59: }
60:
61: private static void createProducerAndSendAMessage() throws JMSException {
62: MessageProducer producer = session.createProducer(destination);
63: producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
64: TextMessage message = session.createTextMessage("Hello World!");
65: System.out.println("Sending message: " + message.getText());
66: producer.send(message);
67: }
68:
69: private static void createConsumerAndReceiveAMessage() throws JMSException, InterruptedException {
70: connection = connectionFactory.createConnection();
71: connection.start();
72: MessageConsumer consumer = session.createConsumer(destination);
73: MyConsumer myConsumer = new MyConsumer();
74: connection.setExceptionListener(myConsumer);
75: consumer.setMessageListener(myConsumer);
76: }
77:
78: private static class MyConsumer implements MessageListener, ExceptionListener {
79:
80: synchronized public void onException(JMSException ex) {
81: System.out.println("JMS Exception occured. Shutting down client.");
82: System.exit(1);
83: }
84:
85: public void onMessage(Message message) {
86: if (message instanceof TextMessage) {
87: TextMessage textMessage = (TextMessage) message;
88: try {
89: System.out.println("Received message: " + textMessage.getText());
90: } catch (JMSException ex) {
91: System.out.println("Error reading message: " + ex);
92: }
93: } else {
94: System.out.println("Received: " + message);
95: }
96: }
97: }
98:}
Technorati Tags: ActiveMQ Asynchronität Messaging
