Activemq Java Example Topic Essays

A Simple Example of Synchronous Message Receives

This section describes the sending and receiving clients in an example that uses the method to consume messages synchronously. This section then explains how to compile, package, and run the clients using the GlassFish Server.

The following sections describe the steps in creating and running the example.

Writing the Clients for the Synchronous Receive Example

    The sending client, , performs the following steps:

  1. Injects resources for a connection factory, queue, and topic:

    @Resource(lookup = "jms/ConnectionFactory") private static ConnectionFactory connectionFactory; @Resource(lookup = "jms/Queue")private static Queue queue; @Resource(lookup = "jms/Topic")private static Topic topic;
  2. Retrieves and verifies command-line arguments that specify the destination type and the number of arguments:

    final int NUM_MSGS; String destType = args[0]; System.out.println("Destination type is " + destType); if ( ! ( destType.equals("queue") || destType.equals("topic") ) ) { System.err.println("Argument must be \”queue\” or " + "\”topic\”"); System.exit(1); } if (args.length == 2){ NUM_MSGS = (new Integer(args[1])).intValue(); } else { NUM_MSGS = 1; }
  3. Assigns either the queue or topic to a destination object, based on the specified destination type:

    Destination dest = null; try { if (destType.equals("queue")) { dest = (Destination) queue; } else { dest = (Destination) topic; } } catch (Exception e) { System.err.println("Error setting destination: " + e.toString()); e.printStackTrace(); System.exit(1); }
  4. Creates a and a :

    Connection connection = connectionFactory.createConnection(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  5. Creates a and a :

    MessageProducer producer = session.createProducer(dest); TextMessage message = session.createTextMessage();
  6. Sends one or more messages to the destination:

    for (int i = 0; i < NUM_MSGS; i++) { message.setText("This is message " + (i + 1) + " from producer"); System.out.println("Sending message: " + message.getText()); producer.send(message); }
  7. Sends an empty control message to indicate the end of the message stream:

    producer.send(session.createMessage());

    Sending an empty message of no specified type is a convenient way to indicate to the consumer that the final message has arrived.

  8. Closes the connection in a block, automatically closing the session and :

    } finally { if (connection != null) { try { connection.close(); } catch (JMSException e) { } } }

    The receiving client, , performs the following steps:

  1. Injects resources for a connection factory, queue, and topic.

  2. Assigns either the queue or topic to a destination object, based on the specified destination type.

  3. Creates a and a .

  4. Creates a :

    consumer = session.createConsumer(dest);
  5. Starts the connection, causing message delivery to begin:

    connection.start();
  6. Receives the messages sent to the destination until the end-of-message-stream control message is received:

    while (true) { Message m = consumer.receive(1); if (m != null) { if (m instanceof TextMessage) { message = (TextMessage) m; System.out.println("Reading message: " + message.getText()); } else { break; } } }

    Because the control message is not a , the receiving client terminates the loop and stops receiving messages after the control message arrives.

  7. Closes the connection in a block, automatically closing the session and .

The method can be used in several ways to perform a synchronous receive. If you specify no arguments or an argument of , the method blocks indefinitely until a message arrives:

Message m = consumer.receive(); Message m = consumer.receive(0);

For a simple client, this may not matter. But if you do not want your application to consume system resources unnecessarily, use a timed synchronous receive. Do one of the following:

  • Call the method with a timeout argument greater than :

    Message m = consumer.receive(1); // 1 millisecond
  • Call the method, which receives a message only if one is available:

    Message m = consumer.receiveNoWait();

The client uses an indefinite loop to receive messages, calling with a timeout argument. Calling would have the same effect.

Starting the JMS Provider

When you use the GlassFish Server, your JMS provider is the GlassFish Server. Start the server as described in Starting and Stopping the GlassFish Server.

To Create JMS Administered Objects for the Synchronous Receive Example

Creating the JMS administered objects for this section involves the following:

  • Creating a connection factory

  • Creating two destination resources

If you built and ran the example in Chapter 17, A Message-Driven Bean Example and did not delete the resources afterward, you need to create only the topic resource.

You can create these objects using the Ant tool. To create all the resources, follow these steps.

  1. In a terminal window, go to the directory:

  2. To create all the resources, type the following command:

    To create only the topic resource, type the following command:

    These Ant targets use the command to create the connection factory and the destination resources.

  3. To verify that the resources have been created, use the following command:


    The output looks like this:


    jms/Queue jms/Topic jms/ConnectionFactory Command list-jms-resources executed successfully.

Building, Packaging, Deploying, and Running the Clients for the Synchronous Receive Example

To run these examples using the GlassFish Server, package each one in an application client JAR file. The application client JAR file requires a manifest file, located in the directory for each example, along with the file.

The file for each example contains Ant targets that compile and package the example. The targets place the file for the example in the directory. Then the targets use the command to package the class file and the manifest file in an application client JAR file.

Because the examples use the common interfaces, you can run them using either a queue or a topic.

To Build and Package the Clients for the Synchronous Receive Example Using NetBeans IDE

  1. In NetBeans IDE, select File->Open Project.

  2. In the Open Project dialog, navigate to:


  3. Select the folder.

  4. Select the Open as Main Project check box.

  5. Click Open Project.

  6. In the Projects tab, right-click the project and select Build.

  7. Select the folder.

  8. Select the Open as Main Project check box.

  9. Click Open Project.

  10. In the Projects tab, right-click the project and select Build.

To Deploy and Run the Clients for the Synchronous Receive Example Using NetBeans IDE

  1. Deploy and run the example:

    1. Right-click the project and select Properties.

    2. Select Run from the Categories tree.

    3. In the Arguments field, type the following:

    4. Click OK.

    5. Right-click the project and select Run.

      The output of the program looks like this (along with some application client container output):


      Destination type is queue Sending message: This is message 1 from producer Sending message: This is message 2 from producer Sending message: This is message 3 from producer

      The messages are now in the queue, waiting to be received.


      Note –

      When you run an application client, there is usually a noticeable delay between the first two application client container messages and the remainder of the output.


  2. Now deploy and run the example:

    1. Right-click the project and select Properties.

    2. Select Run from the Categories tree.

    3. In the Arguments field, type the following:

    4. Click OK.

    5. Right-click the project and select Run.

      The output of the program looks like this (along with some application client container output):


      Destination type is queue Reading message: This is message 1 from producer Reading message: This is message 2 from producer Reading message: This is message 3 from producer
  3. Now try running the programs in the opposite order. Right-click the project and select Run.

    The Output pane displays the destination type and then appears to hang, waiting for messages.

  4. Right-click the project and select Run.

    The Output pane shows the output of both programs, in two different tabs.

  5. Now run the example using a topic instead of a queue.

    1. Right-click the project and select Properties.

    2. Select Run from the Categories tree.

    3. In the Arguments field, type the following:

    4. Click OK.

    5. Right-click the project and select Run.

      The output looks like this (along with some application client container output):


      Destination type is topic Sending message: This is message 1 from producer Sending message: This is message 2 from producer Sending message: This is message 3 from producer
  6. Now run the example using the topic.

    1. Right-click the project and select Properties.

    2. Select Run from the Categories tree.

    3. In the Arguments field, type the following:

    4. Click OK.

    5. Right-click the project and select Run.

      The result, however, is different. Because you are using a topic, messages that were sent before you started the consumer cannot be received. (See Publish/Subscribe Messaging Domain, for details.) Instead of receiving the messages, the program appears to hang.

  7. Run the example again. Right-click the project and select Run.

    Now the example receives the messages:


    Destination type is topic Reading message: This is message 1 from producer Reading message: This is message 2 from producer Reading message: This is message 3 from producer

To Build and Package the Clients for the Synchronous Receive Example Using Ant

  1. In a terminal window, go to the directory:

  2. Type the following command:

  3. In a terminal window, go to the directory:

  4. Type the following command:

    The targets place the application client JAR file in the directory for each example.

To Deploy and Run the Clients for the Synchronous Receive Example Using Ant and the Command

You can run the clients using the command. The file for each project includes a target that deploys the client and then retrieves the client stubs that the command uses. Each of the clients takes one or more command-line arguments: a destination type and, for , a number of messages.

To build, deploy, and run the and examples using Ant and the command, follow these steps.

To run the clients, you need two terminal windows.

  1. In a terminal window, go to the directory:

  2. Deploy the client JAR file to the GlassFish Server, then retrieve the client stubs:

    Ignore the message that states that the application is deployed at a URL.

  3. Run the program, sending three messages to the queue:


    The output of the program looks like this (along with some application client container output):


    Destination type is queue Sending message: This is message 1 from producer Sending message: This is message 2 from producer Sending message: This is message 3 from producer

    The messages are now in the queue, waiting to be received.


    Note –

    When you run an application client, there is usually a noticeable delay between the first two application client container messages and the remainder of the output.


  4. In the same window, go to the directory:

  5. Deploy the client JAR file to the GlassFish Server, then retrieve the client stubs:

    Ignore the message that states that the application is deployed at a URL.

  6. Run the client, specifying the queue:


    The output of the client looks like this (along with some application client container output):


    Destination type is queue Reading message: This is message 1 from producer Reading message: This is message 2 from producer Reading message: This is message 3 from producer
  7. Now try running the clients in the opposite order. Run the client:


    The client displays the destination type and then appears to hang, waiting for messages.

  8. In a different terminal window, run the client.


    When the messages have been sent, the client receives them and exits.

  9. Now run the client using a topic instead of a queue:


    The output of the client looks like this (along with some application client container output):


    Destination type is topic Sending message: This is message 1 from producer Sending message: This is message 2 from producer Sending message: This is message 3 from producer
  10. Now run the client using the topic:


    The result, however, is different. Because you are using a topic, messages that were sent before you started the consumer cannot be received. (See Publish/Subscribe Messaging Domain, for details.) Instead of receiving the messages, the client appears to hang.

  11. Run the client again.

    Now the client receives the messages (along with some application client container output):


    Destination type is topic Reading message: This is message 1 from producer Reading message: This is message 2 from producer Reading message: This is message 3 from producer
  • © 2010, Oracle Corporation and/or its affiliates

The following is a very simple JMS application with multiple, concurrent, consumers and producers. See the Initial Configuration guide for details on how to setup your classpath correctly.

Things you might do after running this example:

  • Setup a broker instead of using the org.activemq.broker.impl.Main class directly
  • Use JNDI to lookup a javax.jms.ConnectionFactory rather than creating ActiveMQConnectionFactory directly.
  • Implement the javax.jms.MessageListener interface rather than calling consumer.receive()
  • Use transactional sessions
  • Use a Topic rather than a queue

The point of this example is to show you the basic code required to use JMS in a straightforward way. The Consumers and Producers could very easy be on completely different machines or in different processes.

import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.Connection; import javax.jms.DeliveryMode; import javax.jms.Destination; import javax.jms.ExceptionListener; import javax.jms.JMSException; import javax.jms.Message; import javax.jms.MessageConsumer; import javax.jms.MessageProducer; import javax.jms.Session; import javax.jms.TextMessage; /** * Hello world! */ public class App { public static void main(String[] args) throws Exception { thread(new HelloWorldProducer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); Thread.sleep(1000); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); Thread.sleep(1000); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldProducer(), false); Thread.sleep(1000); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldConsumer(), false); thread(new HelloWorldProducer(), false); } public static void thread(Runnable runnable, boolean daemon) { Thread brokerThread = new Thread(runnable); brokerThread.setDaemon(daemon); brokerThread.start(); } public static class HelloWorldProducer implements Runnable { public void run() { try { // Create a ConnectionFactory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost"); // Create a Connection Connection connection = connectionFactory.createConnection(); connection.start(); // Create a Session Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageProducer from the Session to the Topic or Queue MessageProducer producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // Create a messages String text = "Hello world! From: " + Thread.currentThread().getName() + " : " + this.hashCode(); TextMessage message = session.createTextMessage(text); // Tell the producer to send the message System.out.println("Sent message: "+ message.hashCode() + " : " + Thread.currentThread().getName()); producer.send(message); // Clean up session.close(); connection.close(); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } } public static class HelloWorldConsumer implements Runnable, ExceptionListener { public void run() { try { // Create a ConnectionFactory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost"); // Create a Connection Connection connection = connectionFactory.createConnection(); connection.start(); connection.setExceptionListener(this); // Create a Session Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageConsumer from the Session to the Topic or Queue MessageConsumer consumer = session.createConsumer(destination); // Wait for a message Message message = consumer.receive(1000); if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; String text = textMessage.getText(); System.out.println("Received: " + text); } else { System.out.println("Received: " + message); } consumer.close(); session.close(); connection.close(); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } public synchronized void onException(JMSException ex) { System.out.println("JMS Exception occured. Shutting down client."); } } }

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *