ActiveMQ点对点的消息发送案例

时间：2022-08-15 05:41:13

公司最近会用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

相关文章