JMS 初步

电玩女神 2022-04-14 03:10 153阅读 0赞

# JMS 初步 #

## 简介 ##

### 百度百科上对JMS的简介： ###

JMS即Java消息服务（Java Message Service）应用程序接口，
    是一个Java平台中关于面向消息中间件（MOM）的API，
    用于在两个应用程序之间，或分布式系统中发送消息，
    进行异步通信。
    Java消息服务是一个与具体平台无关的API，
    绝大多数MOM提供商都对JMS提供支持。

### 百度百科上对ActiveMQ的简介： ###

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。
    ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，
    尽管JMS规范出台已经是很久的事情了，
    但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

### JMS具有两种通信模式： ###

Point-to-Point Messaging Domain （点对点）
    Publish/Subscribe Messaging Domain （发布/订阅模式）

### 比如： ###

你要和某个服务器通信，你得写套接字，连接后双方各send和recv消息。
    而你用JMS，就是向消息服务器（暂且称为服务器）发送一个消息，
    而接收方，只要从这个服务器取下消息即可，不用去写套接字啦。
    JMS是一套规范，各个厂商有不同的实现。比如ActiveMQ，WebsphereMQ等。
    这里称之为消息中间件。
    用了消息中间件，通信的双方都是通过这个中间件来收取消息。
    （这样中间件就通用了，不用再去为某两个程序专门写套接字，有了新程序还得再写一遍）。
    这里的发送的消息可以是字符串，可以是对象。

## 模型图 ##

（图片来自[https://www.cnblogs.com/jaycekon/p/6220200.html][https_www.cnblogs.com_jaycekon_p_6220200.html]）

### 点对点 ###

![RgCR0Vq.png][]

### 发布/订阅模式 ###

![JtJlO4Z.png][]

## 准备 ##

1.  下载activemq程序包，解压，启动。
2.  打开activemq的web管理员控制台 [http://localhost:8161/admin/index.jsp][http_localhost_8161_admin_index.jsp]
3.  添加activemq的jar包到程序中。
4.  编写代码，JavaWeb项目

### 点对点 ###

**Servlet容器的context.xml文件添加JNDI配置**

<Resource name="queue/connectionFactory" auth="Container"
    	type="org.apache.activemq.ActiveMQConnectionFactory"
    	description="JMS Connection Factory"
    	factory="org.apache.activemq.jndi.JNDIReferenceFactory"
    	brokerURL="tcp://localhost:61616" brokerName="LocalActiveMQBroker" />
    
    
    <Resource name="queue/queue0" auth="Container"
    	type="org.apache.activemq.command.ActiveMQQueue"
    	description="My Queue"
    	factory="org.apache.activemq.jndi.JNDIReferenceFactory"
    	physicalName="TomcatQueue" />
    
    
    @WebServlet("/send")
    public class QueueSend extends HttpServlet {
    	private static final long serialVersionUID = 1L;
    
    	protected void doGet(HttpServletRequest request, HttpServletResponse response)
    			throws ServletException, IOException {
    		Writer writer = response.getWriter();
    
    		try {
    			InitialContext context = new InitialContext();
    			QueueConnectionFactory conFactory = (QueueConnectionFactory) context
    					.lookup("java:comp/env/queue/connectionFactory");
    
    			QueueConnection queConn = conFactory.createQueueConnection();
    			queConn.start();
    
    			QueueSession queSession = queConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
    			QueueSender queSender = queSession.createSender((Queue) context.lookup("java:comp/env/queue/queue0"));
    			queSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    
    			TextMessage message = queSession.createTextMessage("Hello World");
    			queSender.send(message);
    
    			queSender.close();
    			queSession.close();
    			queConn.close();
    
    			writer.write("Message Sent: " + message.getText());
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }

--------------------

@WebServlet("/receive")
    public class QueueReceive extends HttpServlet {
    	private static final long serialVersionUID = 1L;
    
    	protected void doGet(HttpServletRequest request, HttpServletResponse response)
    			throws ServletException, IOException {
    		Writer writer = response.getWriter();
    
    		try {
    			InitialContext context = new InitialContext();
    			QueueConnectionFactory conFactory = (QueueConnectionFactory) context
    					.lookup("java:comp/env/queue/connectionFactory");
    
    			QueueConnection queConn = conFactory.createQueueConnection();
    			queConn.start();
    
    			QueueSession queSession = queConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    			QueueReceiver queReceiver = queSession.createReceiver((Queue) context.lookup("java:comp/env/queue/queue0"));
    
    			TextMessage message = (TextMessage) queReceiver.receive();
    			String msg = message.getText();
    
    			queReceiver.close();
    			queSession.close();
    			queConn.close();
    
    			writer.write("Message Received: " + msg);
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }

--------------------

先访问send，在访问receive，注意观察控制台变化。

### 发布/订阅模式 ###

import java.io.IOException;
    
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.MessageProducer;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    import javax.servlet.ServletException;
    import javax.servlet.annotation.WebServlet;
    import javax.servlet.http.HttpServlet;
    import javax.servlet.http.HttpServletRequest;
    import javax.servlet.http.HttpServletResponse;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    	@WebServlet("/topic_send")
    	public class TopicSend extends HttpServlet {
    		private static final long serialVersionUID = 1L;
    	
    		@Override
    		protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
    			try {
    				ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    				Connection connection = connectionFactory.createConnection();
    				connection.start();
    	
    				Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    				// 也可以在JNDI中配置
    				Topic topic = session.createTopic("test-topic");
    				MessageProducer producer = session.createProducer(topic);
    				TextMessage textMessage = session.createTextMessage("hello!test-topic");
    				producer.send(textMessage);
    	
    				producer.close();
    				session.close();
    				connection.close();
    	
    				resp.getWriter().write("Send message : " + textMessage.getText());
    			} catch (Exception e) {
    				e.printStackTrace();
    			}
    		}
    	
    	}

--------------------

import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.MessageListener;
    import javax.jms.Session;
    import javax.jms.TextMessage;
    import javax.jms.Topic;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    public class TopicReceive {
    	public static void main(String[] args) {
    		try {
    			ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:61616");
    			Connection connection = connectionFactory.createConnection();
    			connection.start();
    
    			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    			Topic topic = session.createTopic("test-topic");
    
    			MessageConsumer consumer = session.createConsumer(topic);
    
    			consumer.setMessageListener(new MessageListener() {
    				@Override
    				public void onMessage(Message message) {
    					if (message instanceof TextMessage) {
    						TextMessage textMessage = (TextMessage) message;
    						try {
    							System.out.println("Message received : " + textMessage.getText());
    						} catch (JMSException e) {
    							e.printStackTrace();
    						}
    					}
    				}
    			});
    
    			// 程序等待接收用户消息
    			System.in.read();
    
    			consumer.close();
    			session.close();
    			connection.close();
    		} catch (Exception e) {
    			e.printStackTrace();
    		}
    	}
    }

--------------------

先启动receive，为了启动监听器，再访问send，之一观察控制台的变化。

[https_www.cnblogs.com_jaycekon_p_6220200.html]: https://www.cnblogs.com/jaycekon/p/6220200.html
[RgCR0Vq.png]: https://i.imgur.com/RgCR0Vq.png
[JtJlO4Z.png]: https://i.imgur.com/JtJlO4Z.png
[http_localhost_8161_admin_index.jsp]: http://localhost:8161/admin/index.jsp