ActiveMQ 学习教程

ccgxl123 贡献于2014-03-28

作者 paff  创建于2008-12-26 08:45:00   修改者Administrator  修改于2014-02-12 03:15:52字数9092

文档摘要: ActiveMQ是Apache出品,最流行的,能力强劲的开源消息总线。选择ActiveMQ作为JMS的入门学习中间件,是因为其拥有以下优点1.多种语言和协议编写客户端。语言:Java,C,C++,C#,Ruby,Perl,Python,PHP。应用协议:OpenWire,StompREST,WSNotification,XMPP,AMQP2.完全支持JMS1.1和J2EE1.4规范(持久化,XA消息,事务)
关键词:

ActiveMQ学习教程(一)——安装与示例 背景: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下载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代码 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 LAST_PRICES = new Hashtable(); 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代码 110 import javax.jms.Connection; 111 import javax.jms.ConnectionFactory; 112 import javax.jms.Destination; 113 import javax.jms.JMSException; 114 import javax.jms.MessageConsumer; 115 import javax.jms.Session; 116 117 import org.apache.activemq.ActiveMQConnectionFactory; 118 119 public class Consumer { 120 121 private static String brokerURL = "tcp://localhost:61616"; 122 private static transient ConnectionFactory factory; 123 private transient Connection connection; 124 private transient Session session; 125 126 public Consumer() throws JMSException { 127 factory = new ActiveMQConnectionFactory(brokerURL); 128 connection = factory.createConnection(); 129 connection.start(); 130 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 131 } 132 133 public void close() throws JMSException { 134 if (connection != null) { 135 connection.close(); 136 } 137 } 138 139 public static void main(String[] args) throws JMSException { 140 Consumer consumer = new Consumer(); 141 for (String stock : args) { 142 Destination destination = consumer.getSession().createTopic("STOCKS." + stock); 143 MessageConsumer messageConsumer = consumer.getSession().createConsumer(destination); 144 messageConsumer.setMessageListener(new Listener()); 145 } 146 } 147 148 public Session getSession() { 149 return session; 150 } 151 152 } Listener.java Java代码 153 import java.text.DecimalFormat; 154 155 import javax.jms.MapMessage; 156 import javax.jms.Message; 157 import javax.jms.MessageListener; 158 159 public class Listener implements MessageListener { 160 161 public void onMessage(Message message) { 162 try { 163 MapMessage map = (MapMessage)message; 164 String stock = map.getString("stock"); 165 double price = map.getDouble("price"); 166 double offer = map.getDouble("offer"); 167 boolean up = map.getBoolean("up"); 168 DecimalFormat df = new DecimalFormat( "#,###,###,##0.00" ); 169 System.out.println(stock + "\t" + df.format(price) + "\t" + df.format(offer) + "\t" + (up?"up":"down")); 170 } catch (Exception e) { 171 e.printStackTrace(); 172 } 173 } 174 175 } 先运行Consumer.java, 输入参数ORCL,然后运行Publisher.java,输入参数ORCL, 就可以看到Publisher在发送消息,Consumer在接收消息了。 (不知道怎么在ECLIPSE里带参数运行程序的,请自行GOOGLE。) 好了,MQ的安装与第一个示例程序的介绍就到此为止了。 ActiveMQ学习教程(二)——概念解释 这一节对JMS API中的一些重要概念进行一下说明。 JMS API的主要概念如一下: JMS client: 纯Java写的用于发送和接收消息的应用程序 Non-JMS client: 用JMS提供的本地用户API来发送和接收消息的应用程序 JMS producer: 创建和发送JMS消息的客户端 JMS consumer: 接收和处理JMS消息的客户端 JMS provider: 完全用Java写的JMS接口的实现 JMS message: JMS消息 JMS domains: 两种发送方式,包括点对点(point-to-point)和发布/订阅(publish/subscribe) Administered objects: 预定义的包含JMS provider的配置数据,客户端一般通过JNDI调用 Connection factory: 客户端用Connection Factory创建Connection来连接JMS provider Destination: 用于记录消息的目的地址或接收地址。 本节只列出提纲,详细说明,请查看附件。 下面对两种JMS domains进行一下说明: 1.publish/subscribe 在这种模式下,MQ服务器中的客户端可订阅自己感兴趣的TOPIC,当其它客户端向MQ服务器发送一个TOPIC且该TOPIC是自己订阅过的TOPIC,则接收该消息。上一节的例子就是用了这个模式 2.point-to-point 在这种模式下,消息是点对点发送的,一个QUEUE只会发送到一个客户端。 ActiveMQ学习教程(三)——连接配置 这节介绍MQ的连接配置 在MQ目录/conf/activemq-demo.xml中可看到如下配置 Xml代码 176 177 179 180 181 182 可以看到在MQ中可配置个连接,并且还可以配置多种协议的连接。 MQ中支持的网络协议有TCP、NIO、UDP、SSL、HTTP(S)、VM等。 下面对其中常用的几种进行一下说明。 TCP: MQ默认使用的协议,开放性、高效性、可靠性是其主要特点,这个协议可满足一般应用的大部分需求,因此在项目中还是比较常用的。第一节中的例子用到的就是TCP协议。 NIO: NIO的概念是在JAVASE1.4首先提出来的,这是一种新的JAVA IO模型,是对底层IO的一种封装,让程序员们可以更方便的对IO进行操作。 这种协议下,网络消息的传播效率非常的高。因此,当系统存在高用户量、高并发或网络堵塞时,可考虑用此协议。 配置示例如下: Xml代码 183 SSL: 安全套接层协议,这种协议比较重要,因此本教程将对SSL协议连接方式作重点介绍。 在MQ目录/conf/activemq.xml文件中修改以下配置 Xml代码 184 185 186 187 188 189 broker.ks是MQ的默认证书,默认密码为password 启动activemq.bat 可发现MQ已经以SSL方式启动,端口为61617 将第一节中Consumer.java与Publisher.java的以下代码 Java代码 190 protected static String brokerURL = "tcp://localhost:61616"; 修改为 Java代码 191 protected static String brokerURL = "ssl://localhost:61617"; 在Consumer.java和Publisher的VM arguments 里输入 -Djavax.net.ssl.keyStore=MQ目录\conf\client.ks -Djavax.net.ssl.keyStorePassword=password -Djavax.net.ssl.trustStore=MQ目录\conf\client.ts 设置Consumer.java和Publisher的Program arguments为 ORCL 启动Consumer.java和Publisher,将看到和第一节示例一样的结果,只不过这里的连接协议改为了SSL。 由于本例中用到的证书是MQ自带的,正式环境中不可能用默认证书,否则SSL将形同虚设。因此,我们必须生成自己的证书,以保证其安全性。 在命令行中输入 keytool -genkey -alias broker -keyalg RSA -keystore mybroker.ks 按提示输入密码(本例子的密码为test123)和其它信息,可以看到在当前目录下生成了一个mybroker.ks文件,接着在命令行输入keytool -export -alias broker -keystore mybroker.ks -file mybroker_cert,输入刚才的密码test123,生成mybroker_cert。 接成生成客户端keystore,输入keytool -genkey -alias client -keyalg RSA -keystore myclient.ks,输入客户端密码和其它信息,生成myclient.ks,接着输入keytool -import -alias broker -keystore myclient.ts -file mybroker_cert,生成myclient.ts。 证书生成的工作就完成了。 接着把刚刚生成的证书复制到MQ目录\conf下,将activemq.xml的sslContext节点修改为 Xml代码 192 193 194 重启MQ,修改Consumer.java和Publisher.java和VM参数为 -Djavax.net.ssl.keyStore=MQ目录\conf\myclient.ks -Djavax.net.ssl.keyStorePassword=test123 -Djavax.net.ssl.trustStore=MQ目录\conf\myclient.ts 启动Consumer.java和Publisher.java 到此,我们把我们自己生成的证书用到了SSL协议上。 其它诸如HTTP(S)/UDP(不可靠性)/VM(用于虚拟机内应用程序间的通信)等协议,在实际中用的并不多,本教程就不介绍了。

下载文档到电脑,查找使用更方便

文档的实际排版效果,会与网站的显示效果略有不同!!

需要 5 金币 [ 分享文档获得金币 ] 0 人已下载

下载文档