ActiveMQ学习教程（一）——安装与示例

心已赠人 2022-07-27 17:59 125阅读 0赞

### [IT程序员开发必备-各类资源下载清单][IT_-][，史上最全IT资源，个人收藏总结！][IT_-] ###

背景：ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。  
选择ActiveMQ作为JMS的入门学习中间件，是因为其拥有以下优点

*    
    1.多种语言和协议编写客户端。语言: Java, C, C++, C\#, Ruby, Perl, Python, PHP。应用协议: OpenWire,Stomp REST,WS Notification,XMPP,AMQP  
    2.完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)  
    3.对Spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性  
    4.完全支持JMS1.1和J2EE 1.4规范 (持久化,XA消息,事务)  
    5.通过了常见J2EE服务器(如 Geronimo,JBoss 4, GlassFish,WebLogic)的测试,其中通过JCA 1.5 resource adaptors的配置,可以让ActiveMQ可以自动的部署到任何兼容J2EE 1.4 商业服务器上  
    6.支持多种传送协议:in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA  
    7.从设计上保证了高性能的集群,客户端-服务器,点对点  
    8.支持Ajax  
    9.支持与Axis的整合  
    10.可以很容易得调用内嵌JMS provider,进行测试

学会了ActiveMQ之后，其它供应商的MQ也可以在短时间内快速上手。  
  
安装：  
ActiveMQ（本文简称MQ）要求JDK1.5以上，推荐1.6以上版本。还没安装JDK的朋友，请先安装，在此不赘诉了。  
安装完JDK后，从 [ http://activemq.apache.org/download.html][http_activemq.apache.org_download.html]下载MQ的最新版本，本教程使用版本为5.5。  
解压后，可以看到MQ目录下有以下文件和目录

*    
    activemq-all-5.5.0.jar:所有MQ JAR包的集合，用于用户系统调用  
    bin:其中包含MQ的启动脚本  
    conf:包含MQ的所有配置文件  
    data:日志文件及持久性消息数据  
    example:MQ的示例  
    lib:MQ运行所需的所有Lib  
    webapps:MQ的Web控制台及一些相关的DEMO

启动MQ：  
双击bin目录下的activemq.bat文件即可启动MQ  
  
  
第一个示例：  
  
新建一个JAVA工程，引用activemq-all-5.5.0.jar，SLFAPI其及对应版本LOG4J的JAR包（懒的上网找的到附件里下载）  
  
Publisher.java

Java代码   ![收藏代码][icon_star.png]

1.  import java.util.Hashtable;  
2.  import java.util.Map;  
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.MapMessage;  
9.  import javax.jms.Message;  
10. import javax.jms.MessageProducer;  
11. import javax.jms.Session;  
12.   
13. import org.apache.activemq.ActiveMQConnectionFactory;  
14. import org.apache.activemq.command.ActiveMQMapMessage;  
15.   
16. public class Publisher \{  
17.       
18.     protected int MAX\_DELTA\_PERCENT = 1;  
19.     protected Map<String, Double> LAST\_PRICES = new Hashtable<String, Double>();  
20.     protected static int count = 10;  
21.     protected static int total;  
22.       
23.     protected static String brokerURL = "tcp://localhost:61616";  
24.     protected static transient ConnectionFactory factory;  
25.     protected transient Connection connection;  
26.     protected transient Session session;  
27.     protected transient MessageProducer producer;  
28.       
29.     public Publisher() throws JMSException \{  
30.         factory = new ActiveMQConnectionFactory(brokerURL);  
31.         connection = factory.createConnection();  
32.         try \{  
33.         connection.start();  
34.         \} catch (JMSException jmse) \{  
35.             connection.close();  
36.             throw jmse;  
37.         \}  
38.         session = connection.createSession(false, Session.AUTO\_ACKNOWLEDGE);  
39.         producer = session.createProducer(null);  
40.     \}  
41.       
42.     public void close() throws JMSException \{  
43.         if (connection != null) \{  
44.             connection.close();  
45.         \}  
46.     \}  
47.       
48.     public static void main(String\[\] args) throws JMSException \{  
49.         Publisher publisher = new Publisher();  
50.         while (total < 1000) \{  
51.             for (int i = 0; i < count; i++) \{  
52.                 publisher.sendMessage(args);  
53.             \}  
54.             total += count;  
55.             System.out.println("Published '" + count + "' of '" + total + "' price messages");  
56.             try \{  
57.               Thread.sleep(1000);  
58.             \} catch (InterruptedException x) \{  
59.             \}  
60.         \}  
61.         publisher.close();  
62.     \}  
63.   
64.     protected void sendMessage(String\[\] stocks) throws JMSException \{  
65.         int idx = 0;  
66.         while (true) \{  
67.             idx = (int)Math.round(stocks.length \* Math.random());  
68.             if (idx < stocks.length) \{  
69.                 break;  
70.             \}  
71.         \}  
72.         String stock = stocks\[idx\];  
73.         Destination destination = session.createTopic("STOCKS." + stock);  
74.         Message message = createStockMessage(stock, session);  
75.         System.out.println("Sending: " + ((ActiveMQMapMessage)message).getContentMap() + " on destination: " + destination);  
76.         producer.send(destination, message);  
77.     \}  
78.   
79.     protected Message createStockMessage(String stock, Session session) throws JMSException \{  
80.         Double value = LAST\_PRICES.get(stock);  
81.         if (value == null) \{  
82.             value = new Double(Math.random() \* 100);  
83.         \}  
84.   
85.         // lets mutate the value by some percentage  
86.         double oldPrice = value.doubleValue();  
87.         value = new Double(mutatePrice(oldPrice));  
88.         LAST\_PRICES.put(stock, value);  
89.         double price = value.doubleValue();  
90.   
91.         double offer = price \* 1.001;  
92.   
93.         boolean up = (price > oldPrice);  
94.   
95.         MapMessage message = session.createMapMessage();  
96.         message.setString("stock", stock);  
97.         message.setDouble("price", price);  
98.         message.setDouble("offer", offer);  
99.         message.setBoolean("up", up);  
100.         return message;  
101.     \}  
102.   
103.     protected double mutatePrice(double price) \{  
104.         double percentChange = (2 \* Math.random() \* MAX\_DELTA\_PERCENT) - MAX\_DELTA\_PERCENT;  
105.   
106.         return price \* (100 + percentChange) / 100;  
107.     \}  
108.   
109. \}

Consumer.java

Java代码   ![收藏代码][icon_star.png]

1.  import javax.jms.Connection;  
2.  import javax.jms.ConnectionFactory;  
3.  import javax.jms.Destination;  
4.  import javax.jms.JMSException;  
5.  import javax.jms.MessageConsumer;  
6.  import javax.jms.Session;  
7.    
8.  import org.apache.activemq.ActiveMQConnectionFactory;  
9.    
10. public class Consumer \{  
11.   
12.     private static String brokerURL = "tcp://localhost:61616";  
13.     private static transient ConnectionFactory factory;  
14.     private transient Connection connection;  
15.     private transient Session session;  
16.       
17.     public Consumer() throws JMSException \{  
18.         factory = new ActiveMQConnectionFactory(brokerURL);  
19.         connection = factory.createConnection();  
20.         connection.start();  
21.         session = connection.createSession(false, Session.AUTO\_ACKNOWLEDGE);  
22.     \}  
23.       
24.     public void close() throws JMSException \{  
25.         if (connection != null) \{  
26.             connection.close();  
27.         \}  
28.     \}      
29.       
30.     public static void main(String\[\] args) throws JMSException \{  
31.         Consumer consumer = new Consumer();  
32.         for (String stock : args) \{  
33.             Destination destination = consumer.getSession().createTopic("STOCKS." + stock);  
34.             MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination);  
35.             messageConsumer.setMessageListener(new Listener());  
36.         \}  
37.     \}  
38.       
39.     public Session getSession() \{  
40.         return session;  
41.     \}  
42.   
43. \}

Listener.java

Java代码   ![收藏代码][icon_star.png]

1.  import java.text.DecimalFormat;  
2.    
3.  import javax.jms.MapMessage;  
4.  import javax.jms.Message;  
5.  import javax.jms.MessageListener;  
6.    
7.  public class Listener implements MessageListener \{  
8.    
9.      public void onMessage(Message message) \{  
10.         try \{  
11.             MapMessage map = (MapMessage)message;  
12.             String stock = map.getString("stock");  
13.             double price = map.getDouble("price");  
14.             double offer = map.getDouble("offer");  
15.             boolean up = map.getBoolean("up");  
16.             DecimalFormat df = new DecimalFormat( "\#,\#\#\#,\#\#\#,\#\#0.00" );  
17.             System.out.println(stock + "\\t" + df.format(price) + "\\t" + df.format(offer) + "\\t" + (up?"up":"down"));  
18.         \} catch (Exception e) \{  
19.             e.printStackTrace();  
20.         \}  
21.     \}  
22.   
23. \}

先运行Consumer.java, 输入参数ORCL，然后运行Publisher.java，输入参数ORCL，  
就可以看到Publisher在发送消息，Consumer在接收消息了。  
（不知道怎么在ECLIPSE里带参数运行程序的，请自行GOOGLE。）  
好了，MQ的安装与第一个示例程序的介绍就到此为止了。

[IT_-]: http://blog.csdn.net/xh16319/article/details/16356913
[http_activemq.apache.org_download.html]: http://activemq.apache.org/download.html
[icon_star.png]: /images/20220727/dc46bc69a18f4ca9b28db94e03d42427.png