【ActiveMQ】ActiveMQ在Windows的安装,以及点对点的消息发送案例

公司最近会用MQ对某些业务进行处理,所以,这次我下载了apache-activemq-5.12.0-bin把玩下。

基于练习方便需要,使用Windows的版本。

参考的优秀文章:

activemq的几种基本通信方式总结

Winsdows下安装

下载后得到zip文件,解压后用命令行启动即可使用,有两个问题需注意。

  1. 启动的文件根据不同的系统类型,也分为32位、64位,请选择自身系统的启动文件
  2. 目前的apache-activemq-5.12.0版本需要1.7的JDK运行
D:\green\apache-activemq-5.12.\bin\win64\activemq.bat

启动后,可见如下类型的日志,即正常启动:

C:\Users\Administrator>D:\green\apache-activemq-5.12.\bin\win64\activemq.bat
wrapper | --> Wrapper Started as Console
wrapper | Launching a JVM...
jvm | Wrapper (Version 3.2.) http://wrapper.tanukisoftware.org
jvm | Copyright - Tanuki Software, Inc. All Rights Reserved.
jvm |
jvm | Java Runtime: Oracle Corporation 1.7.0_71 D:\green\JDK\jdk1..0_71\jr
e7
jvm | Heap sizes: current=61440k free=55845k max=932352k
jvm | JVM args: -Dactivemq.home=../.. -Dactivemq.base=../.. -Djavax.net
.ssl.keyStorePassword=password -Djavax.net.ssl.trustStorePassword=password -Djav
ax.net.ssl.keyStore=../../conf/broker.ks -Djavax.net.ssl.trustStore=../../conf/b
roker.ts -Dcom.sun.management.jmxremote -Dorg.apache.activemq.UseDedicatedTaskRu
nner=true -Djava.util.logging.config.file=logging.properties -Dactivemq.conf=../
../conf -Dactivemq.data=../../data -Djava.security.auth.login.config=../../conf/
login.config -Xmx1024m -Djava.library.path=../../bin/win64 -Dwrapper.key=XfRgHYr
WI1FG6qgT -Dwrapper.port= -Dwrapper.jvm.port.min= -Dwrapper.jvm.port.m
ax= -Dwrapper.pid= -Dwrapper.version=3.2. -Dwrapper.native_library=wra
pper -Dwrapper.cpu.timeout= -Dwrapper.jvmid=
jvm | Extensions classpath:
jvm | [..\..\lib,..\..\lib\camel,..\..\lib\optional,..\..\lib\web,..\..\l
ib\extra]
jvm | ACTIVEMQ_HOME: ..\..
jvm | ACTIVEMQ_BASE: ..\..
jvm | ACTIVEMQ_CONF: ..\..\conf
jvm | ACTIVEMQ_DATA: ..\..\data
jvm | Loading message broker from: xbean:activemq.xml
jvm | INFO | Refreshing org.apache.activemq.xbean.XBeanBrokerFactory$@3b7
94c60: startup date [Tue Aug :: CST ]; root of context hierarchy
jvm | INFO | PListStore:[D:\green\apache-activemq-5.12.\bin\win64\..\..\d
ata\localhost\tmp_storage] started
jvm | INFO | Using Persistence Adapter: KahaDBPersistenceAdapter[D:\green\
apache-activemq-5.12.\bin\win64\..\..\data\kahadb]
jvm | INFO | KahaDB is version
jvm | INFO | Recovering from the journal @:
jvm | INFO | Recovery replayed operations from the journal in 0.02 secon
ds.
jvm | INFO | Apache ActiveMQ 5.12. (localhost, ID:USER-20141117SF--
-:) is starting
jvm | INFO | Listening for connections at: tcp://USER-20141117SF:61616?max
imumConnections=&wireFormat.maxFrameSize=
jvm | INFO | Connector openwire started
jvm | INFO | Listening for connections at: amqp://USER-20141117SF:5672?max
imumConnections=&wireFormat.maxFrameSize=
jvm | INFO | Connector amqp started
jvm | INFO | Listening for connections at: stomp://USER-20141117SF:61613?m
aximumConnections=&wireFormat.maxFrameSize=
jvm | INFO | Connector stomp started
jvm | INFO | Listening for connections at: mqtt://USER-20141117SF:1883?max
imumConnections=&wireFormat.maxFrameSize=
jvm | INFO | Connector mqtt started
jvm | {}
jvm | INFO | Listening for connections at ws://USER-20141117SF:61614?maxim
umConnections=&wireFormat.maxFrameSize=
jvm | INFO | Connector ws started
jvm | INFO | Apache ActiveMQ 5.12. (localhost, ID:USER-20141117SF--
-:) started
jvm | INFO | For help or more information please see: http://activemq.apac
he.org
jvm | INFO | ActiveMQ WebConsole available at http://0.0.0.0:8161/
jvm | INFO | ActiveMQ Jolokia REST API available at http://0.0.0.0:8161/ap
i/jolokia/
jvm | INFO | Initializing Spring FrameworkServlet 'dispatcher'
jvm | INFO | jolokia-agent: No access restrictor found at classpath:/jolok
ia-access.xml, access to all MBeans is allowed

默认情况下, activemq服务监听61616端口,如此端口被占用,会报错,请释放后再试。

控制台

http://127.0.0.1:8161/admin/

默认账户/密码:admin/admin

基于JMS的点对点队列,实现异步队列发送短信

在点对点模式(P2P或point to point)中,消息发送方称为生产者(Producer),消息接收方称为消费者(Consumer)。而在发布订阅模式中,则分别为发布者、订阅者。

系统需要给客户发送短信,一般来说,我们都是通过调用3大运营商的短信接口进行发送短信的。

而在某些时候,比如春节,发送短信的人数特别多,所以接口变得堵塞,发送短信需时较久。我们不希望因发送短信堵塞应用程序,因此,我们想把发送短信的部分放到MQ,从而实现异步发送短信。(当然,在这种情况下,我们不关心发送短信是否成功。即,即使发送短信失败,在系统的界面并不提示用户)。

以下只是学习例子,并不适应实际案例。

消费者持续连接MQ服务器,接收消息,并实现发送短信的业务

package com.nicchagil.activemq.study.No002队列异步发送短信案例;

import java.util.HashMap;
import java.util.concurrent.TimeUnit; import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.ObjectMessage;
import javax.jms.Session; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; public class SMSConsumer { public static void main(String[] args) throws JMSException {
ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD,
"tcp://127.0.0.1:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("SendSMS");
MessageConsumer consumer = session.createConsumer(destination); while (true) {
ObjectMessage message = (ObjectMessage)consumer.receive();
if (message != null) {
HashMap<String, String> messageMap = (HashMap<String, String>)message.getObject();
sendSMS(messageMap.get("mobile"), messageMap.get("message"));
}
}
} public static void sendSMS(String mobile, String message) {
System.out.println("Send SMS to : " + mobile + ", content : " + message); /* 模拟在繁忙时间发送短信需时15秒 */
try {
TimeUnit.SECONDS.sleep(15);
} catch (InterruptedException e) {
e.printStackTrace();
} System.out.println("Done");
} }

两个生产者,用于发送消息,调用消费者发送短信。两个生产者意在测试,他们的不同点仅在于短信文本。

package com.nicchagil.activemq.study.No002队列异步发送短信案例;

import java.util.HashMap;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; public class SMSProducer1 { public static void main(String[] args) throws JMSException {
/* 连接 */
ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
Connection connection = factory.createConnection();
connection.start(); Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 会话
Destination destination = session.createQueue("SendSMS"); // 队列 /* 生产者 */
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); /* 发送消息 */
HashMap<String, String> map = new HashMap<String, String>();
map.put("mobile", "13000000000");
map.put("message", "2016,新年快乐!");
ObjectMessage message = session.createObjectMessage(map);
producer.send(message);
session.commit();
System.out.println("sent...");
} }
package com.nicchagil.activemq.study.No002队列异步发送短信案例;

import java.util.HashMap;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; public class SMSProducer2 { public static void main(String[] args) throws JMSException {
/* 连接 */
ConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
Connection connection = factory.createConnection();
connection.start(); Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE); // 会话
Destination destination = session.createQueue("SendSMS"); // 队列 /* 生产者 */
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); /* 发送消息 */
HashMap<String, String> map = new HashMap<String, String>();
map.put("mobile", "13000000000");
map.put("message", "猴年快乐哦!");
ObjectMessage message = session.createObjectMessage(map);
producer.send(message);
session.commit();
System.out.println("sent...");
} }

运行后消费者的日志(Main方法运行顺序:SMSConsumer > SMSProducer1 > SMSProducer2)

Send SMS to : , content : ,新年快乐!
Done
Send SMS to : , content : 猴年快乐哦!
Done
上一篇:【227】◀▶ IDL 其他常用函数


下一篇:Android Afinal框架