The term “JMS client” refers to Java components or applications that use the JMS API and a JMS provider to send and receive messages.<\/p>\n
JMS supports two styles of messaging: the point\u2212to\u2212point and publis\u2212and\u2212subscribe messaging styles. Before a client can use a JMS provider to send and receive messages, the client must decide which messaging style it wants to use.<\/p>\n
A client can consume a message synchronously or asynchronously.<\/p>\n
In this article, we will see several examples of JMS Clients.<\/p>\n
In order to send and receive JMS messages to and from a JMS message broker, we need to include the message service library. In this example we are using activeMq so our pom.xml will have dependencies related to spring as well as activeMq.<\/p>\n
pom.xml:<\/em><\/span><\/p>\n JMS is a specification and not an actual product. A JMS provider such as ActiveMQ, IBM, Progress Software, or even Sun provides a messaging product that implements the specification. In our examples, we will be using ActiveMQ as JMS Provider. Getting started with ActiveMQ isn\u2019t difficult. You simply need to start up the broker and make sure that it\u2019s capable of accepting connections and sending messages.<\/p>\n In the below example, the broker is started as a server listening on port 61616. The JMS Clients willing to connect to the broker will be using the TCP protocol (tcp:\/\/localhost:61616). Since the broker and JMS clients are running in the same machine, we have used localhost.<\/p>\n BrokerLauncher:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n For a client to interact with the JMS provider to needs to get hold of a connection to the broker and a connection represents a logical connection to the JMS provider. In order to obtain this connection each JMS provider provides a connection factory. There are two types of connection factories: one for point\u2212to\u2212point and another for publish\u2212and\u2212subscribe. Based on the desired messaging style, the client obtains the appropriate connection factory and connects to the JMS provider.<\/p>\n In case of ActiveMQ, it provides one ConnectionFactory and internally it implements both In our next sections, we will show how to create a connection factory, create a new connection and session, create message producers and consumers which we will then use to send and receive messages. First let\u2019s look at a message producer client. We will use point\u2212to\u2212point messaging style.<\/p>\n We will first obtain a connection factory, which we will then use to create a connection.<\/p>\n Next, use the connection object to create a queue session.<\/p>\n The session object obtained is used to create a producer that will be used to send a message. When the producer is created, it is told which queue to send the messages to.<\/p>\n Next, we create messages and send in a loop.<\/p>\n Finally, we send message ‘END’ to indicate the client that we have sent the last message.<\/p>\n Finally, close the session and the connections.<\/p>\n JmsProducerClient:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n The consumer is very similar to the producer client. It also needs a connection factory, the connection, the session, and the same queue. In this client program, however, the session is used to create a consumer instead of a producer. This consumer is told which queue to consume messages from when it is created.<\/p>\n To actually receive a message, the client calls the receive method as follows:<\/p>\n As you can see the consumer receives in an infinite loop. If it receives ‘END’, it comes out of the loop. The receive method can be used in several ways to perform a synchronous receive. If you specify no arguments or an argument of 0, the method blocks indefinitely until a message arrives.<\/p>\n JmsSyncReceiveClientExample:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n This section describes how to consume messages asynchronously. It uses a message listener in order to consume messages asynchronously. In order to make sure the asynchronous consumer doesn’t run indefinitely, it calls ConsumerMessageListener:<\/em><\/span><\/p>\n Just like the synchronous client, it also creates a JmsAsyncReceiveQueueClientExample:<\/em><\/span>[ulp id=’meNHaG2axN7AsWAw’]<\/p>\n Output:<\/em><\/span><\/p>\n Let’s now look into a client that uses publish\u2212and\u2212subscribe message style. It\u2019s not very different than using the point\u2212to\u2212point style. It also needs a connection factory, connection and a session.<\/p>\n When a publisher sends a message, there may be more than one customer interested in such messages. Publisher broadcasts the message to JMS destination called \u2018topic\u2019. There may be more than one consumer subscribed to the topic. All the active clients subscribed to the topic will receive message and there is no need for the subscriber to poll for the messages. Every active subscriber receives its own copy of each message published to the topic. In this example, we will look into durable subscriber.<\/p>\n So what is a durable subscriber?<\/p>\n When a publisher publishes messages for an inactive subscriber, the messages are persisted and delivered when the subscriber reconnects. For durable subscribers to a topic, each consumer gets a copy of the message. While a durable subscriber is disconnected from the JMS server, it is the responsibility of the server to store messages the subscriber misses. When the durable subscriber reconnects, the message server sends it all the unexpired messages that accumulated.<\/p>\n The JmsSubscriberClientExample:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n We have seen examples of synchronous and asynchronous client. We have also seen clients using point-to-point and publisher-subscriber style of messaging. Let’s now see how to view the messages sent to a consumer without actually consuming them.<\/p>\n In this example, we will show you how to browse through the pending messages in the queue using As you can see In the below example, we create a producer and post a bunch of messages to a queue. Next we create a consumer. In order to browse, we create a QueueBrowser object and navigate through the messages. Finally, we call JmsBrowseQueueClientExample:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n Suppose each message that a JMS provider delivers to a consumer need not be acknowledged and is left up to the client consuming the messages to decide when to acknowledge, then we need to create as session by passing in If a message is not acknowledged it may be redelivered to the consumer by the JMS provider. The client acknowledges a message by calling the acknowledge method on it.<\/p>\n Acknowledging one message actually acknowledges all messages that the session has consumed.<\/p>\n JmsClientAckExample:<\/em><\/span><\/p>\n Output:<\/em><\/span><\/p>\n This was an example about JMS Clients.<\/p>\n<project xmlns=\"http:\/\/maven.apache.org\/POM\/4.0.0\" xmlns:xsi=\"http:\/\/www.w3.org\/2001\/XMLSchema-instance\"\n\txsi:schemaLocation=\"http:\/\/maven.apache.org\/POM\/4.0.0 http:\/\/maven.apache.org\/xsd\/maven-4.0.0.xsd\">\n\t<modelVersion>4.0.0<\/modelVersion>\n\t<groupId>com.javacodegeeks.jms<\/groupId>\n\t<artifactId>springJmsQueue<\/artifactId>\n\t<version>0.0.1-SNAPSHOT<\/version>\n\t<dependencies>\n\t\t<dependency>\n\t\t\t<groupId>org.apache.activemq<\/groupId>\n\t\t\t<artifactId>activemq-all<\/artifactId>\n\t\t\t<version>5.12.0<\/version>\n\t\t<\/dependency>\n\t<\/dependencies>\t\n<\/project>\n<\/pre>\n<\/p>\n
2. Starting the JMS Provider<\/h2>\n
package com.javacodegeeks.jms;\n\nimport java.net.URI;\nimport java.net.URISyntaxException;\n\nimport org.apache.activemq.broker.BrokerFactory;\nimport org.apache.activemq.broker.BrokerService;\n\npublic class BrokerLauncher {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tBrokerService broker = BrokerFactory.createBroker(new URI(\n\t\t\t\t\"broker:(tcp:\/\/localhost:61616)\"));\n\t\tbroker.start();\t\t\n\t}\n}\n<\/pre>\n INFO | JMX consoles can connect to service:jmx:rmi:\/\/\/jndi\/rmi:\/\/localhost:1099\/jmxrmi\n INFO | PListStore:[C:\\javacodegeeks_ws\\jmsClientExample\\activemq-data\\localhost\\tmp_storage] started\n INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[C:\\javacodegeeks_ws\\jmsClientExample\\activemq-data\\localhost\\KahaDB]\n INFO | KahaDB is version 6\n INFO | Recovering from the journal @1:15633\n INFO | Recovery replayed 62 operations from the journal in 0.016 seconds.\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-57384-1447857883439-0:1) is starting\n INFO | Listening for connections at: tcp:\/\/127.0.0.1:61616\n INFO | Connector tcp:\/\/127.0.0.1:61616 started\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-57384-1447857883439-0:1) started\n INFO | For help or more information please see: http:\/\/activemq.apache.org\n WARN | Store limit is 102400 mb (current store usage is 0 mb). The data directory: C:\\javacodegeeks_ws\\jmsClientExample\\activemq-data\\localhost\\KahaDB only has 37428 mb of usable space - resetting to maximum available disk space: 37428 mb\n WARN | Temporary Store limit is 51200 mb, whilst the temporary data directory: C:\\javacodegeeks_ws\\jmsClientExample\\activemq-data\\localhost\\tmp_storage only has 37428 mb of usable space - resetting to maximum available 37428 mb.\n<\/pre>\n
3. ConnectionFactory<\/h2>\n
QueueConnectionFactory<\/code> and TopicConnectionFactory<\/code>.<\/p>\nConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"tcp:\/\/localhost:61616\");\n<\/pre>\n
4. JMS Producer Client<\/h2>\n
\/\/ Producer\nConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"tcp:\/\/localhost:61616\");\nconnection = connectionFactory.createConnection();\n<\/pre>\n
Session session = connection.createSession(false,\n\t\t\t\t\tSession.AUTO_ACKNOWLEDGE);\nQueue queue = session.createQueue(\"customerQueue\");\n<\/pre>\n
MessageProducer producer = session.createProducer(queue);\n<\/pre>\n
for (int i = 0; i < 10; i++) {\n String payload = task + i;\n Message msg = session.createTextMessage(payload);\n System.out.println(\"Sending text '\" + payload + \"'\");\n producer.send(msg);\n}\n<\/pre>\nproducer.send(session.createTextMessage(\"END\"));\n<\/pre>\n
try {\n ...\n session.close();\n} finally {\n if (connection != null) {\n\tconnection.close();\n }\n}\n<\/pre>\npackage com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.Message;\nimport javax.jms.MessageProducer;\nimport javax.jms.Queue;\nimport javax.jms.Session;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsProducerQueueClient {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tConnection connection = null;\n\t\ttry {\n\t\t\t\/\/ Producer\n\t\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"tcp:\/\/localhost:61616\");\n\t\t\tconnection = connectionFactory.createConnection();\n\t\t\tSession session = connection.createSession(false,\n\t\t\t\t\tSession.AUTO_ACKNOWLEDGE);\n\t\t\tQueue queue = session.createQueue(\"customerQueue\");\n\t\t\tMessageProducer producer = session.createProducer(queue);\n\t\t\tString task = \"Task\";\n\t\t\tfor (int i = 0; i < 10; i++) {\n\t\t\t\tString payload = task + i;\n\t\t\t\tMessage msg = session.createTextMessage(payload);\n\t\t\t\tSystem.out.println(\"Sending text '\" + payload + \"'\");\n\t\t\t\tproducer.send(msg);\n\t\t\t}\n\t\t\tproducer.send(session.createTextMessage(\"END\"));\n\t\t\tsession.close();\n\t\t} finally {\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\n\t}\n}\n<\/pre>\nSending text 'Task0'\nSending text 'Task1'\nSending text 'Task2'\nSending text 'Task3'\nSending text 'Task4'\nSending text 'Task5'\nSending text 'Task6'\nSending text 'Task7'\nSending text 'Task8'\nSending text 'Task9'\n<\/pre>\n
5. JMS Consumer Client<\/h2>\n
\/\/ Consumer\nMessageConsumer consumer = session.createConsumer(queue);\nconnection.start();\n<\/pre>\n
while (true) {\n TextMessage textMsg = (TextMessage) consumer.receive();\n System.out.println(textMsg);\n System.out.println(\"Received: \" + textMsg.getText());\n if (textMsg.getText().equals(\"END\")) {\n break;\n }\n}\t\n<\/pre>\npackage com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.MessageConsumer;\nimport javax.jms.Queue;\nimport javax.jms.Session;\nimport javax.jms.TextMessage;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsSyncReceiveClientExample {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tConnection connection = null;\n\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\"tcp:\/\/localhost:61616\");\n\t\tconnection = connectionFactory.createConnection();\n\t\tSession session = connection.createSession(false,\n\t\t\t\tSession.AUTO_ACKNOWLEDGE);\n\t\ttry {\n\t\t\tQueue queue = session.createQueue(\"customerQueue\");\n\n\t\t\t\/\/ Consumer\n\t\t\tMessageConsumer consumer = session.createConsumer(queue);\n\t\t\tconnection.start();\n\t\t\twhile (true) {\n\t\t\t\tTextMessage textMsg = (TextMessage) consumer.receive();\n\t\t\t\tSystem.out.println(textMsg);\n\t\t\t\tSystem.out.println(\"Received: \" + textMsg.getText());\n\t\t\t\tif (textMsg.getText().equals(\"END\")) {\n\t\t\t\t\tbreak;\n\t\t\t\t}\n\t\t\t}\t\t\t\n\t\t} finally {\n\t\t\tif (session != null) {\n\t\t\t\tsession.close();\n\t\t\t}\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\n\t}\n}\n<\/pre>\nActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931210, arrival = 0, brokerInTime = 1447857931212, brokerOutTime = 1447857977960, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@6e3c1e69, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task0}\nReceived: Task0\nActiveMQTextMessage {commandId = 6, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:2, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931222, arrival = 0, brokerInTime = 1447857931222, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@1888ff2c, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task1}\nReceived: Task1\nActiveMQTextMessage {commandId = 7, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:3, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931224, arrival = 0, brokerInTime = 1447857931225, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@35851384, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task2}\nReceived: Task2\nActiveMQTextMessage {commandId = 8, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:4, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931227, arrival = 0, brokerInTime = 1447857931227, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@649d209a, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task3}\nReceived: Task3\nActiveMQTextMessage {commandId = 9, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:5, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931230, arrival = 0, brokerInTime = 1447857931230, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@6adca536, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task4}\nReceived: Task4\nActiveMQTextMessage {commandId = 10, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:6, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931232, arrival = 0, brokerInTime = 1447857931233, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@357246de, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task5}\nReceived: Task5\nActiveMQTextMessage {commandId = 11, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:7, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931235, arrival = 0, brokerInTime = 1447857931235, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@28f67ac7, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task6}\nReceived: Task6\nActiveMQTextMessage {commandId = 12, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:8, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931238, arrival = 0, brokerInTime = 1447857931238, brokerOutTime = 1447857977961, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@256216b3, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task7}\nReceived: Task7\nActiveMQTextMessage {commandId = 13, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:9, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931240, arrival = 0, brokerInTime = 1447857931241, brokerOutTime = 1447857977962, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@2a18f23c, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task8}\nReceived: Task8\nActiveMQTextMessage {commandId = 14, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:10, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931243, arrival = 0, brokerInTime = 1447857931243, brokerOutTime = 1447857977962, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@d7b1517, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task9}\nReceived: Task9\nActiveMQTextMessage {commandId = 15, responseRequired = true, messageId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1:11, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-57438-1447857931037-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447857931246, arrival = 0, brokerInTime = 1447857931246, brokerOutTime = 1447857977962, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@16c0663d, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = END}\nReceived: END\n<\/pre>\n6. JMS Asynchronous Client Example<\/h2>\n
countDown()<\/code> on latch when the message received is ‘END’.<\/p>\npackage com.javacodegeeks.jms;\n\nimport javax.jms.JMSException;\nimport javax.jms.Message;\nimport javax.jms.MessageListener;\nimport javax.jms.TextMessage;\n\npublic class ConsumerMessageListener implements MessageListener {\n\tprivate String consumerName;\n\tprivate JmsAsyncReceiveQueueClientExample asyncReceiveQueueClientExample;\n\t\n\tpublic ConsumerMessageListener(String consumerName) {\n\t\tthis.consumerName = consumerName;\n\t}\n\n\tpublic void onMessage(Message message) {\n\t\tTextMessage textMessage = (TextMessage) message;\n\t\ttry {\n\t\t\tSystem.out.println(consumerName + \" received \" + textMessage.getText());\n\t\t\tif (\"END\".equals(textMessage.getText())) {\n\t\t\t\tasyncReceiveQueueClientExample.latchCountDown();\n\t\t\t}\n\t\t} catch (JMSException e) {\t\t\t\n\t\t\te.printStackTrace();\n\t\t}\n\t}\n\n\tpublic void setAsyncReceiveQueueClientExample(\n\t\t\tJmsAsyncReceiveQueueClientExample asyncReceiveQueueClientExample) {\n\t\tthis.asyncReceiveQueueClientExample = asyncReceiveQueueClientExample;\n\t}\t\n}\n<\/pre>\nConnection<\/code>, a Session<\/code>, a MessageConsumer<\/code> and creates an instance of the MessageListener<\/code> class and then registers it as the message listener for the MessageConsumer<\/code>.<\/p>\nMessageConsumer consumer = session.createConsumer(queue);\nConsumerMessageListener consumerListener = new ConsumerMessageListener(\n\t\t\t\t\t\"Customer\");\nconsumer.setMessageListener(consumerListener);\n<\/pre>\n
package com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\nimport java.util.concurrent.CountDownLatch;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.JMSException;\nimport javax.jms.MessageConsumer;\nimport javax.jms.Queue;\nimport javax.jms.Session;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsAsyncReceiveQueueClientExample {\n\tprivate CountDownLatch latch = new CountDownLatch(1);\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tJmsAsyncReceiveQueueClientExample asyncReceiveClient = new JmsAsyncReceiveQueueClientExample();\n\t\tasyncReceiveClient.receiveMessages();\n\t}\n\n\tpublic void receiveMessages() throws JMSException, InterruptedException {\n\t\tConnection connection = null;\n\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\"tcp:\/\/localhost:61616\");\n\t\tconnection = connectionFactory.createConnection();\n\t\tSession session = connection.createSession(false,\n\t\t\t\tSession.AUTO_ACKNOWLEDGE);\n\t\ttry {\n\t\t\tQueue queue = session.createQueue(\"customerQueue\");\n\n\t\t\t\/\/ Consumer\n\t\t\tMessageConsumer consumer = session.createConsumer(queue);\n\t\t\tConsumerMessageListener consumerListener = new ConsumerMessageListener(\n\t\t\t\t\t\"Customer\");\n\t\t\tconsumer.setMessageListener(consumerListener);\n\t\t\tconsumerListener.setAsyncReceiveQueueClientExample(this);\n\t\t\n\t\t\tconnection.start();\n\t\t\tlatch.await();\n\t\t} finally {\n\t\t\tif (session != null) {\n\t\t\t\tsession.close();\n\t\t\t}\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\t\t\n\t}\n\n\tpublic void latchCountDown() {\n\t\tlatch.countDown();\n\t}\n}\n<\/pre>\nCustomer received Task0\nCustomer received Task1\nCustomer received Task2\nCustomer received Task3\nCustomer received Task4\nCustomer received Task5\nCustomer received Task6\nCustomer received Task7\nCustomer received Task8\nCustomer received Task9\nCustomer received END\n<\/pre>\n
7. JMS Subscriber Client<\/h2>\n
createDurableSubscriber()<\/code> method takes two parameters: a topic name, and a subscription name. A durable subscription\u2019s uniqueness is defined by the client ID and the subscription name.<\/p>\npackage com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.Message;\nimport javax.jms.MessageConsumer;\nimport javax.jms.MessageProducer;\nimport javax.jms.Session;\nimport javax.jms.TextMessage;\nimport javax.jms.Topic;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsSubscriberClientExample {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tConnection connection = null;\n\t\ttry {\n\t\t\t\/\/ Producer\n\t\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"tcp:\/\/localhost:61616\");\n\t\t\tconnection = connectionFactory.createConnection();\n\t\t\tconnection.setClientID(\"DurabilityTest\");\n\t\t\tSession session = connection.createSession(false,\n\t\t\t\t\tSession.AUTO_ACKNOWLEDGE);\n\t\t\tTopic topic = session.createTopic(\"customerTopic\");\n\n\t\t\t\/\/ Publish\n\t\t\tString payload = \"Task\";\n\t\t\tTextMessage msg = session.createTextMessage(payload);\n\t\t\tMessageProducer publisher = session.createProducer(topic);\n\t\t\tSystem.out.println(\"Sending text '\" + payload + \"'\");\n\t\t\tpublisher.send(msg, javax.jms.DeliveryMode.PERSISTENT, javax.jms.Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);\n\n\t\t\t\/\/ Consumer1 subscribes to customerTopic\n\t\t\tMessageConsumer consumer1 = session.createDurableSubscriber(topic, \"consumer1\", \"\", false);\t\t\n\n\t\t\t\/\/ Consumer2 subscribes to customerTopic\n\t\t\tMessageConsumer consumer2 = session.createDurableSubscriber(topic, \"consumer2\", \"\", false);\t\n\n\t\t\tconnection.start();\n\t\t\t\n\t\t\tmsg = (TextMessage) consumer1.receive();\n\t\t\tSystem.out.println(\"Consumer1 receives \" + msg.getText());\n\t\t\t\n\t\t\t\n\t\t\tmsg = (TextMessage) consumer2.receive();\n\t\t\tSystem.out.println(\"Consumer2 receives \" + msg.getText());\n\n\t\t\tsession.close();\n\t\t} finally {\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\n\t}\n}\n<\/pre>\nSending text 'Task'\nConsumer1 receives Task\nConsumer2 receives Task\n<\/pre>\n
8. Client Program for the Queue Browser<\/h2>\n
QueueBrowser<\/code> object. Since we can browse through the messages without actually consuming them, this is very unique and important feature to point-to-point messaging. We create the QueueBrowser object using the below statement on session object.<\/p>\nQueueBrowser browser = session.createBrowser(queue);\n<\/pre>\n
createBrowser()<\/code> takes the Queue object that we are interested to browse. In order to enumerate through the messages, we will call QueueBrowser.getEnumeration()<\/code>. When we are done with the browser we should close it using <codeQueueBrowser.close().<\/p>\nconsumer.receive()<\/code> to receive one of the messages from queue.<\/p>\npackage com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\nimport java.util.Enumeration;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.MessageConsumer;\nimport javax.jms.Queue;\nimport javax.jms.QueueBrowser;\nimport javax.jms.Session;\nimport javax.jms.TextMessage;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsBrowseQueueClientExample {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tConnection connection = null;\n\t\ttry {\n\t\t\t\/\/ Producer\n\t\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"tcp:\/\/localhost:61616\");\n\t\t\tconnection = connectionFactory.createConnection();\n\t\t\tSession session = connection.createSession(false,\n\t\t\t\t\tSession.AUTO_ACKNOWLEDGE);\n\t\t\tQueue queue = session.createQueue(\"customerQueue\");\n\n\t\t\tMessageConsumer consumer = session.createConsumer(queue);\n\t\t\tconnection.start();\n\t\t\t\n\t\t\tSystem.out.println(\"Browse through the elements in queue\");\n\t\t\tQueueBrowser browser = session.createBrowser(queue);\n\t\t\tEnumeration e = browser.getEnumeration();\n\t\t\twhile (e.hasMoreElements()) {\n\t\t\t\tTextMessage message = (TextMessage) e.nextElement();\n\t\t\t\tSystem.out.println(\"Get [\" + message.getText() + \"]\");\n\t\t\t}\n\t\t\tSystem.out.println(\"Done\");\n\t\t\tbrowser.close();\n\t\t\t\n\t\t\tTextMessage textMsg = (TextMessage) consumer.receive();\n\t\t\tSystem.out.println(textMsg);\n\t\t\tSystem.out.println(\"Received: \" + textMsg.getText());\n\t\t\tsession.close();\n\t\t} finally {\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\n\t}\n\n}\n<\/pre>\nBrowse through the elements in queue\nGet [Task0]\nGet [Task1]\nGet [Task2]\nGet [Task3]\nGet [Task4]\nGet [Task5]\nGet [Task6]\nGet [Task7]\nGet [Task8]\nGet [Task9]\nGet [END]\nDone\nActiveMQTextMessage {commandId = 5, responseRequired = true, messageId = ID:INMAA1-L1005-58212-1447859579333-1:1:1:1:1, originalDestination = null, originalTransactionId = null, producerId = ID:INMAA1-L1005-58212-1447859579333-1:1:1:1, destination = queue:\/\/customerQueue, transactionId = null, expiration = 0, timestamp = 1447859579480, arrival = 0, brokerInTime = 1447859579481, brokerOutTime = 1447859586255, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = org.apache.activemq.util.ByteSequence@28864e92, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false, text = Task0}\nReceived: Task0\n<\/pre>\n9. JMS Client Acknowledge<\/h2>\n
Session.CLIENT_ACKNOWLEDGE<\/code> as the second parameter.<\/p>\nSession session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);\n<\/pre>\n
Message.acknowledge();\n<\/pre>\n
package com.javacodegeeks.jms;\n\nimport java.net.URISyntaxException;\n\nimport javax.jms.Connection;\nimport javax.jms.ConnectionFactory;\nimport javax.jms.MessageConsumer;\nimport javax.jms.MessageProducer;\nimport javax.jms.Queue;\nimport javax.jms.Session;\nimport javax.jms.TextMessage;\n\nimport org.apache.activemq.ActiveMQConnectionFactory;\n\npublic class JmsClientAckExample {\n\tpublic static void main(String[] args) throws URISyntaxException, Exception {\n\t\tConnection connection = null;\n\t\ttry {\n\t\t\t\/\/ Producer\n\t\t\tConnectionFactory connectionFactory = new ActiveMQConnectionFactory(\n\t\t\t\t\t\"vm:\/\/localhost?broker.persistent=false\");\n\t\t\tconnection = connectionFactory.createConnection();\n\t\t\tconnection.start();\n\t\t\tSession session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);\n\t Queue queue = session.createQueue(\"SomeQueue\");\n\t MessageProducer producer = session.createProducer(queue);\n\t producer.send(session.createTextMessage(\"Hello\"));\n\n\t MessageConsumer consumer = session.createConsumer(queue);\n\t TextMessage msg = (TextMessage) consumer.receive(1000);\n\t System.out.println(\"Consume: \" + msg.getText());\n\t \n\t \/\/ Reset the session.\n\t session.close();\n\t session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);\n\n\t \/\/ Attempt to Consume the message...\n\t consumer = session.createConsumer(queue);\n\t msg = (TextMessage) consumer.receive(1000);\n\t System.out.println(\"Attempt to consume again. Is message received? \" + (msg != null));\n\t \n\t \/\/acknowledge\n\t msg.acknowledge();\n\n\t\t\t\/\/ Reset the session.\n\t session.close();\n\t session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);\n\n\t \/\/ Attempt to Consume the message...\n\t consumer = session.createConsumer(queue);\n\t msg = (TextMessage) consumer.receive(1000);\n\t System.out.println(\"Attempt to consume again. Is message received? \" + (msg != null));\n\t\t} finally {\n\t\t\tif (connection != null) {\n\t\t\t\tconnection.close();\n\t\t\t}\n\t\t}\n\t}\n}\n<\/pre>\n INFO | Using Persistence Adapter: MemoryPersistenceAdapter\n WARN | Failed to start JMX connector Cannot bind to URL [rmi:\/\/localhost:1099\/jmxrmi]: javax.naming.NameAlreadyBoundException: jmxrmi [Root exception is java.rmi.AlreadyBoundException: jmxrmi]. Will restart management to re-create JMX connector, trying to remedy this issue.\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-53531-1447910509729-0:1) is starting\n WARN | Failed to start JMX connector Cannot bind to URL [rmi:\/\/localhost:1099\/jmxrmi]: javax.naming.NameAlreadyBoundException: jmxrmi [Root exception is java.rmi.AlreadyBoundException: jmxrmi]. Will restart management to re-create JMX connector, trying to remedy this issue.\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-53531-1447910509729-0:1) started\n INFO | For help or more information please see: http:\/\/activemq.apache.org\n WARN | Temporary Store limit is 51200 mb, whilst the temporary data directory: C:\\javacodegeeks_ws\\jmsClientExample\\activemq-data\\localhost\\tmp_storage only has 37178 mb of usable space - resetting to maximum available 37178 mb.\n INFO | Connector vm:\/\/localhost started\nConsume: Hello\nAttempt to consume again. Is message received? true\nAttempt to consume again. Is message received? false\n INFO | Connector vm:\/\/localhost stopped\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-53531-1447910509729-0:1) is shutting down\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-53531-1447910509729-0:1) uptime 1.471 seconds\n INFO | Apache ActiveMQ 5.12.0 (localhost, ID:INMAA1-L1005-53531-1447910509729-0:1) is shutdown\n<\/pre>\n
10. Download the Eclipse Project<\/h2>\n