带有ActiveMQ的JMS

带有ActiveMQ的JMS

JMS是Java Message Service的缩写，它提供了一种以松散耦合，灵活的方式集成应用程序的机制。 JMS以存储和转发的方式跨应用程序异步传递数据。应用程序通过充当中介的MOM（面向消息的中间件）进行通信，而无需直接通信。

JMS体系结构

JMS的主要组件是：

JMS Provider：一种消息传递系统，它实现JMS接口并提供管理和控制功能
客户端：发送或接收JMS消息的Java应用程序。消息发送者称为生产者，而接收者称为消费者
消息：在JMS客户端之间传递信息的对象
受管对象：管理员为使用客户端而创建的预配置JMS对象。

有几种JMS提供程序可用，例如Apache ActiveMQ和OpenMQ。在这里，我使用了Apache ActiveMQ。

在Windows上安装和启动Apache ActiveMQ

下载ActiveMQ Windows二进制分发版
将其提取到所需位置
使用命令提示符将目录更改为ActiveMQ安装文件夹中的bin文件夹，然后运行以下命令以启动ActiveMQ
```
activemq
```

启动ActiveMQ之后，您可以使用http：// localhost：8161 / admin /访问管理控制台并执行管理任务

JMS消息传递模型

JMS具有两种消息传递模型，即点对点消息传递模型和发布者订户消息传递模型。

点对点消息传递模型

生产者将消息发送到JMS提供程序中的指定队列，并且只有一个监听该队列的使用者才能接收该消息。

示例1和示例2几乎相似，唯一的区别是示例1在程序内创建队列，示例2使用jndi.properties文件命名查找和创建队列。

例子1

package com.eviac.blog.jms;import javax.jms.*;
import javax.naming.InitialContext;
import javax.naming.NamingException;import org.apache.log4j.BasicConfigurator;public class Producer {public Producer() throws JMSException, NamingException {// Obtain a JNDI connectionInitialContext jndi = new InitialContext();// Look up a JMS connection factoryConnectionFactory conFactory = (ConnectionFactory) jndi.lookup('connectionFactory');Connection connection;// Getting JMS connection from the server and starting itconnection = conFactory.createConnection();try {connection.start();// JMS messages are sent and received using a Session. We will// create here a non-transactional session object. If you want// to use transactions you should set the first parameter to 'true'Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);Destination destination = (Destination) jndi.lookup('MyQueue');// MessageProducer is used for sending messages (as opposed// to MessageConsumer which is used for receiving them)MessageProducer producer = session.createProducer(destination);// We will send a small text message saying 'Hello World!'TextMessage message = session.createTextMessage('Hello World!');// Here we are sending the message!producer.send(message);System.out.println('Sent message '' + message.getText() + ''');} finally {connection.close();}}public static void main(String[] args) throws JMSException {try {BasicConfigurator.configure();new Producer();} catch (NamingException e) {e.printStackTrace();}}
}

package com.eviac.blog.jms;import javax.jms.*;import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.BasicConfigurator;public class Consumer {// URL of the JMS serverprivate static String url = ActiveMQConnection.DEFAULT_BROKER_URL;// Name of the queue we will receive messages fromprivate static String subject = 'MYQUEUE';public static void main(String[] args) throws JMSException {BasicConfigurator.configure();// Getting JMS connection from the serverConnectionFactory connectionFactory = new ActiveMQConnectionFactory(url);Connection connection = connectionFactory.createConnection();connection.start();// Creating session for seding messagesSession session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);// Getting the queueDestination destination = session.createQueue(subject);// MessageConsumer is used for receiving (consuming) messagesMessageConsumer consumer = session.createConsumer(destination);// Here we receive the message.// By default this call is blocking, which means it will wait// for a message to arrive on the queue.Message message = consumer.receive();// There are many types of Message and TextMessage// is just one of them. Producer sent us a TextMessage// so we must cast to it to get access to its .getText()// method.if (message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;System.out.println('Received message '' + textMessage.getText()+ ''');}connection.close();}
}

jndi.properties

# START SNIPPET: jndijava.naming.factory.initial = org.apache.activemq.jndi.ActiveMQInitialContextFactory# use the following property to configure the default connector
java.naming.provider.url = vm://localhost# use the following property to specify the JNDI name the connection factory
# should appear as. 
#connectionFactoryNames = connectionFactory, queueConnectionFactory, topicConnectionFactry# register some queues in JNDI using the form
# queue.[jndiName] = [physicalName]
queue.MyQueue = example.MyQueue# register some topics in JNDI using the form
# topic.[jndiName] = [physicalName]
topic.MyTopic = example.MyTopic# END SNIPPET: jndi

package com.eviac.blog.jms;import javax.jms.*;
import javax.naming.InitialContext;
import javax.naming.NamingException;import org.apache.log4j.BasicConfigurator;public class Producer {public Producer() throws JMSException, NamingException {// Obtain a JNDI connectionInitialContext jndi = new InitialContext();// Look up a JMS connection factoryConnectionFactory conFactory = (ConnectionFactory) jndi.lookup('connectionFactory');Connection connection;// Getting JMS connection from the server and starting itconnection = conFactory.createConnection();try {connection.start();// JMS messages are sent and received using a Session. We will// create here a non-transactional session object. If you want// to use transactions you should set the first parameter to 'true'Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);Destination destination = (Destination) jndi.lookup('MyQueue');// MessageProducer is used for sending messages (as opposed// to MessageConsumer which is used for receiving them)MessageProducer producer = session.createProducer(destination);// We will send a small text message saying 'Hello World!'TextMessage message = session.createTextMessage('Hello World!');// Here we are sending the message!producer.send(message);System.out.println('Sent message '' + message.getText() + ''');} finally {connection.close();}}public static void main(String[] args) throws JMSException {try {BasicConfigurator.configure();new Producer();} catch (NamingException e) {e.printStackTrace();}}
}

package com.eviac.blog.jms;import javax.jms.*;
import javax.naming.InitialContext;
import javax.naming.NamingException;import org.apache.log4j.BasicConfigurator;public class Consumer {public Consumer() throws NamingException, JMSException {Connection connection;// Obtain a JNDI connectionInitialContext jndi = new InitialContext();// Look up a JMS connection factoryConnectionFactory conFactory = (ConnectionFactory) jndi.lookup('connectionFactory');// Getting JMS connection from the server and starting it// ConnectionFactory connectionFactory = new// ActiveMQConnectionFactory(url);connection = conFactory.createConnection();// // Getting JMS connection from the server// ConnectionFactory connectionFactory = new// ActiveMQConnectionFactory(url);// Connection connection = connectionFactory.createConnection();try {connection.start();// Creating session for seding messagesSession session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);// Getting the queueDestination destination = (Destination) jndi.lookup('MyQueue');// MessageConsumer is used for receiving (consuming) messagesMessageConsumer consumer = session.createConsumer(destination);// Here we receive the message.// By default this call is blocking, which means it will wait// for a message to arrive on the queue.Message message = consumer.receive();// There are many types of Message and TextMessage// is just one of them. Producer sent us a TextMessage// so we must cast to it to get access to its .getText()// method.if (message instanceof TextMessage) {TextMessage textMessage = (TextMessage) message;System.out.println('Received message '' + textMessage.getText()+ ''');}} finally {connection.close();}}public static void main(String[] args) throws JMSException {BasicConfigurator.configure();try {new Consumer();} catch (NamingException e) {// TODO Auto-generated catch blocke.printStackTrace();}}
}

发布者订阅者模型

发布者将消息发布到JMS提供程序中的指定主题，并且订阅该主题的所有订阅者都将收到该消息。请注意，只有活动的订户才能收到该消息。

点对点模型示例

package com.eviac.blog.jms;import javax.jms.*;
import javax.naming.*;import org.apache.log4j.BasicConfigurator;import java.io.BufferedReader;
import java.io.InputStreamReader;public class DemoPublisherSubscriberModel implements javax.jms.MessageListener {private TopicSession pubSession;private TopicPublisher publisher;private TopicConnection connection;/* Establish JMS publisher and subscriber */public DemoPublisherSubscriberModel(String topicName, String username,String password) throws Exception {// Obtain a JNDI connectionInitialContext jndi = new InitialContext();// Look up a JMS connection factoryTopicConnectionFactory conFactory = (TopicConnectionFactory) jndi.lookup('topicConnectionFactry');// Create a JMS connectionconnection = conFactory.createTopicConnection(username, password);// Create JMS session objects for publisher and subscriberpubSession = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);TopicSession subSession = connection.createTopicSession(false,Session.AUTO_ACKNOWLEDGE);// Look up a JMS topicTopic chatTopic = (Topic) jndi.lookup(topicName);// Create a JMS publisher and subscriberpublisher = pubSession.createPublisher(chatTopic);TopicSubscriber subscriber = subSession.createSubscriber(chatTopic);// Set a JMS message listenersubscriber.setMessageListener(this);// Start the JMS connection; allows messages to be deliveredconnection.start();// Create and send message using topic publisherTextMessage message = pubSession.createTextMessage();message.setText(username + ': Howdy Friends!');publisher.publish(message);}/** A client can register a message listener with a consumer. A message* listener is similar to an event listener. Whenever a message arrives at* the destination, the JMS provider delivers the message by calling the* listener's onMessage method, which acts on the contents of the message.*/public void onMessage(Message message) {try {TextMessage textMessage = (TextMessage) message;String text = textMessage.getText();System.out.println(text);} catch (JMSException jmse) {jmse.printStackTrace();}}public static void main(String[] args) {BasicConfigurator.configure();try {if (args.length != 3)System.out.println('Please Provide the topic name,username,password!');DemoPublisherSubscriberModel demo = new DemoPublisherSubscriberModel(args[0], args[1], args[2]);BufferedReader commandLine = new java.io.BufferedReader(new InputStreamReader(System.in));// closes the connection and exit the system when 'exit' enters in// the command linewhile (true) {String s = commandLine.readLine();if (s.equalsIgnoreCase('exit')) {demo.connection.close();System.exit(0);}}} catch (Exception e) {e.printStackTrace();}}
}