JMS开源框架ActiveMQ入门

c#小王子 2022-03-12 487 Java

JMS开源框架ActiveMQ入门

介绍基本的JMS概念与开源的JMS框架ActiveMQ应用，内容涵盖一下几点：

1. 基本的JMS概念

2. JMS的消息模式

3. 介绍ActiveMQ

4. 一个基于ActiveMQ的JMS例子程序

一：JMS基本概念

1. JMS的目标

为企业级的应用提供一种智能的消息系统，JMS定义了一整套的企业级的消息概念与工具，尽可能最小化的Java语言概念去构建最大化企业消息应用。统一已经存在的企业级消息系统功能。

2. 提供者

JMS提供者是指那些完全完成JMS功能与管理功能的JMS消息厂商，理论上JMS提供者完成。

JMS消息产品必须是100%的纯Java语言实现，可以运行在跨平台的架构与操作系统上，当前一些JMS厂商包括IBM,Oracle, JBoss社区 (JBoss Community), Apache 社区(ApacheCommunity)。

3. JMS应用程序, 一个完整的JMS应用应该实现以下功能：

· JMS 客户端 – Java语言开发的接受与发送消息的程序

· 非JMS客户端 – 基于消息系统的本地API实现而不是JMS

· 消息 – 应用程序用来相互交流信息的载体

· 被管理对象–预先配置的JMS对象，JMS管理员创建，被客户端运用。如链接工厂，主题等

· JMS提供者–完成JMS功能与管理功能的消息系统

二：JMS的消息模式

1.点对点的消息模式(Point to Point Messaging)

JMS开源框架ActiveMQ入门

下面的JMS对象在点对点消息模式中是必须的：

a.队列(Queue) – 一个提供者命名的队列对象，客户端将会使用这个命名的队列对象

b.队列链接工厂(QueueConnectionFactory) – 客户端使用队列链接工厂创建链接队列

ConnectionQueue来取得与JMS点对点消息提供者的链接。

c. 链接队列(ConnectionQueue) – 一个活动的链接队列存在在客户端与点对点消息提供者之间，客户用它创建一个或者多个JMS队列会话(QueueSession)

d. 队列会话(QueueSession) – 用来创建队列消息的发送者与接受者(QueueSenderand QueueReceiver)

e.消息发送者(QueueSender 或者MessageProducer)– 发送消息到已经声明的队列

f.消息接受者(QueueReceiver或者MessageConsumer) – 接受已经被发送到指定队列的消息

2.发布订阅模式(publish – subscribe Mode)

a.主题Topic(Destination) – 一个提供者命名的主题对象，客户端将会使用这个命名的主题对象

b.主题链接工厂(TopciConnectionFactory) – 客户端使用主题链接工厂创建链接主题

ConnectionTopic来取得与JMS消息Pub/Sub提供者的链接。

c.链接主题(ConnectionTopic) – 一个活动的链接主题存在发布者与订阅者之间

d.会话(TopicSession) – 用来创建主题消息的发布者与订阅者 (TopicPublisher and TopicSubscribers)

e.消息发送者MessageProducer) – 发送消息到已经声明的主题

f.消息接受者(MessageConsumer) – 接受已经被发送到指定主题的消息

三：介绍ActiveMQ

ActiveMQ是apache社区完成的JMS开源消息组件，客户端支持多种语言调用，包括Java,C++, C#,

Perl, Python等。支持Spring配置集成等。更多信息访问这里：

http://activemq.apache.org/index.html

四：基于ActiveMQ的Publish/subscribe模式Demo程序

消息Broker，JMSprovider

1. import java.net.URI;  
 
2. import java.net.URISyntaxException;  
 
3.  
 
4. import javax.jms.Connection;  
 
5. import javax.jms.ConnectionFactory;  
 
6. import javax.jms.Destination;  
 
7. import javax.jms.JMSException;  
 
8. import javax.jms.MessageProducer;  
 
9. import javax.jms.Session;  
 
10. import javax.jms.TextMessage;  
 
11. import javax.naming.Context;  
 
12. import javax.naming.InitialContext;  
 
13. import javax.naming.NamingException;  
 
14.  
 
15. import org.apache.activemq.broker.BrokerFactory;  
 
16. import org.apache.activemq.broker.BrokerService;  
 
17. import org.apache.commons.logging.Log;  
 
18. import org.apache.commons.logging.LogFactory;  
 
19.  
 
20. /**  
 
21.  * refer to http://activemq.apache.org/jndi-support.html  
 
22.  * http://activemq.apache.org/how-do-i-embed-a-broker-inside-a-connection.html  
 
23.  * @author gloomyfish  
 
24.  *  
 
25.  */ 
 
26. public class PureJMSProducer {  
 
27.       
 
28.     private static final Log LOG = LogFactory.getLog(PureJMSProducer.class);  
 
29.  
 
30.     private PureJMSProducer() {  
 
31.     }  
 
32.  
 
33.     /**  
 
34.      * @param args the destination name to send to and optionally, the number of  
 
35.      *                messages to send  
 
36.      */ 
 
37.     public static void main(String[] args) {  
 
38.         Context jndiContext = null;  
 
39.         ConnectionFactory connectionFactory = null;  
 
40.         Connection connection = null;  
 
41.         Session session = null;  
 
42.         Destination destination = null;  
 
43.         MessageProducer producer = null;  
 
44.         BrokerService broker = null;  
 
45.         final int numMsgs = 10;  
 
46.  
 
47.         /*  
 
48.          * Create a JNDI API InitialContext object  
 
49.          */ 
 
50.         try {  
 
51.             jndiContext = new InitialContext();  
 
52.         } catch (NamingException e) {  
 
53.             LOG.info("Could not create JNDI API context: " + e.toString());  
 
54.             System.exit(1);  
 
55.         }  
 
56.           
 
57.         // create external TCP broker  
 
58.         try {  
 
59.             broker = BrokerFactory.createBroker(new URI("broker:tcp://localhost:61616"));  
 
60.             broker.start();   
 
61.         } catch (URISyntaxException e) {  
 
62.             LOG.info("Could not create broker: " + e.toString());  
 
63.         } catch (Exception e) {  
 
64.             LOG.info("Could not create broker: " + e.toString());  
 
65.         }  
 
66. //        try {  
 
67. //            
 
68. //        }  
 
69.  
 
70.         /*  
 
71.          * Look up connection factory and destination.  
 
72.          */ 
 
73.         try {  
 
74.             connectionFactory = (ConnectionFactory)jndiContext.lookup("ConnectionFactory");  
 
75.             destination = (Destination)jndiContext.lookup("MyTopic");  
 
76.         } catch (NamingException e) {  
 
77.             LOG.info("JNDI API lookup failed: " + e);  
 
78.             System.exit(1);  
 
79.         }  
 
80.           
 
81.         /*  
 
82.          * Create connection. Create session from connection; false means  
 
83.          * session is not transacted. Create sender and text message. Send  
 
84.          * messages, varying text slightly. Send end-of-messages message.  
 
85.          * Finally, close connection.  
 
86.          */ 
 
87.         try {  
 
88.             connection = connectionFactory.createConnection();  
 
89.             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);  
 
90.             producer = session.createProducer(destination);  
 
91.             TextMessage message = session.createTextMessage();  
 
92.             Thread.sleep(3000);  
 
93.             for (int i = 0; i < numMsgs; i++) {  
 
94.                 message.setText("This is message " + (i + 1));  
 
95.                 LOG.info("Sending message: " + message.getText());  
 
96.                 producer.send(message);  
 
97.                 Thread.sleep(3000);  
 
98.             }  
 
99.  
 
100.             /*  
 
101.              * Send a non-text control message indicating end of messages.  
 
102.              */ 
 
103.             producer.send(session.createMessage());  
 
104.         } catch (JMSException e) {  
 
105.             LOG.info("Exception occurred: " + e);  
 
106.         } catch (InterruptedException e) {  
 
107.             LOG.info("Exception occurred: " + e);  
 
108.         } finally {  
 
109.             if (connection != null) {  
 
110.                 try {  
 
111.                     connection.close();  
 
112.                 } catch (JMSException e) {  
 
113.                 }  
 
114.             }  
 
115.         }  
 
116.           
 
117.         // stop the TCP broker  
 
118.         try {  
 
119.             broker.stop();  
 
120.         } catch (Exception e) {  
 
121.             LOG.info("stop the broker failed: " + e);  
 
122.         }  
 
123.     }  
 
}

【下载地址】

百度网盘链接：https://pan.baidu.com/s/1dcq8TW_dp1Bb0_EkYZSHtg

提取码：77rf