ActiveMQ JMS实现消息发送

2023-08-10 22:37:34

一、创建配置消息发送接收目的地、 ActiveMQ中间件地址

JMS_BROKER_URL=failover://(tcp://192.168.1.231:61616)

QUEUE_BUSP_TP_SMS_MESSAGE=busp.tp.sms.message

二、创建消息生产者配置

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"

    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"

    xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:aop="http://www.springframework.org/schema/aop"

    xmlns:jee="http://www.springframework.org/schema/jee"

    xsi:schemaLocation="

        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd

        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.1.xsd

        http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.1.xsd

        http://www.springframework.org/schema/jee http://www.springframework.org/schema/jee/spring-jee-2.5.xsd">

    <!-- load config file -->

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">

        <property name="ignoreUnresolvablePlaceholders" value="true" />

        <property name="locations">

            <list>

                <value>classpath:property/sms-jms.properties</value>

            </list>

        </property>

    </bean>

    <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

        <property name="brokerURL" value="${JMS_BROKER_URL}"/>

    </bean>  

    <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->

        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>

    </bean>

      <!-- Spring提供的JMS工具类，它可以进行消息发送、接收等 -->

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">

        <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->

        <property name="connectionFactory" ref="connectionFactory"/>

    </bean>

    <!-- 发送资金流水-->

    <bean id="sendMessageDestination" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg>

            <value>${QUEUE_BUSP_TP_SMS_MESSAGE}</value>

        </constructor-arg>

    </bean>

</beans>

三、创建生产者并发送消息

package com.busp.tp.sms.sender;

import javax.jms.Destination;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.Session;

import org.apache.log4j.Logger;

import org.springframework.context.support.ClassPathXmlApplicationContext;

import org.springframework.jms.core.JmsTemplate;

import org.springframework.jms.core.MessageCreator;

import com.busp.tp.sms.common.SMSMsg;

/**

 * @ClassName: PushJMSSender

 * @version 1.0

 * @Desc: 向jms发送消息

 * @date 2016年4月21日下午6:01:26

 * @history v1.0

 *

 */

@SuppressWarnings("resource")

public class SMSJMSSender {

    // 日志文件

    private static Logger logger = Logger.getLogger(SMSJMSSender.class);

    // jms template

    private static JmsTemplate jmsTemplate;

    // 目标地址

    private static Destination sendMessageDestination;

    static{

        ClassPathXmlApplicationContext ctx = new ClassPathXmlApplicationContext("classpath:sms-jms-sender.xml");

        ctx.registerShutdownHook();

        sendMessageDestination = (Destination) ctx.getBean("sendMessageDestination");

        jmsTemplate = (JmsTemplate) ctx.getBean("jmsTemplate");

    }

   /**

    *

    * 描述：向JMS推送消息（批量）

    * @date 2016年4月21日下午6:31:48

    * @param pushMsg

    */

    public static void sendJMSMessage(SMSMsg smsMsg){

        try {

            final String result = smsMsg.toJSONString();

            logger.info("向JMS推送消息:" + result);

            jmsTemplate.send(sendMessageDestination, new MessageCreator() {

                public Message createMessage(Session session)

                        throws JMSException {

                    return session.createTextMessage(result);

                }

            });

        } catch (Exception e) {

            logger.error("向JMS推送消息发生了异常",e);

        }

    }

}

四、创建接收者配置

<?xml version="1.0" encoding="UTF-8"?>

<beans  xmlns="http://www.springframework.org/schema/beans"

        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xmlns:aop="http://www.springframework.org/schema/aop"

         xmlns:tx="http://www.springframework.org/schema/tx"

         xmlns:context="http://www.springframework.org/schema/context"

         xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"

         default-lazy-init="false">                                                                                                                                                                                                                            

     <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">

        <property name="ignoreUnresolvablePlaceholders" value="true" />

        <property name="locations">

            <list>

                <value>classpath:property/sms-jms.properties</value>

            </list>

        </property>

    </bean>

     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">

        <property name="brokerURL" value="${JMS_BROKER_URL}"/>

    </bean>

     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->

    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">

        <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->

        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>

    </bean>  

    <!-- ################################### 监听放款信息         Start #####################################-->

    <bean id="smsMessageQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">

        <constructor-arg>

            <value>${QUEUE_BUSP_TP_SMS_MESSAGE}</value>

        </constructor-arg>

    </bean>

    <bean id="smsMessageListener" class="com.busp.tp.sms.listener.SMSMessagelListener"/>

    <bean id="smsMessageContainer"

        class="org.springframework.jms.listener.DefaultMessageListenerContainer">

        <property name="connectionFactory" ref="targetConnectionFactory" />

        <property name="destination" ref="smsMessageQueueDestination" />

        <property name="messageListener" ref="smsMessageListener" />

    </bean>

    <!-- ################################### 监听放款信息        End #############################

</beans>

五、接受者接收消息

package com.busp.tp.sms.listener;

import javax.annotation.Resource;

import javax.jms.JMSException;

import javax.jms.Message;

import javax.jms.MessageListener;

import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.busp.common.util.string.StringUtil;

import com.busp.tp.sms.handler.SMSHandler;

/**

 * @ClassName: PushMessageListener

 * @version 1.0

 * @Desc: TODO

 * @date 2016年4月21日下午4:19:18

 * @history v1.0

 *

 */

public class SMSMessagelListener  implements MessageListener{

    // 日志文件

    private Logger logger = Logger.getLogger(SMSMessagelListener.class);

    @Resource

    private SMSHandler yxtSMSHandler;

    //@Resource

    //private PushMsgHandler androidPushMsgHandler;

    @Override

    public void onMessage(Message message) {

        try {

            TextMessage text = (TextMessage) message;

            final String value = text.getText();

            logger.info("接收到客户端发来的消息："+ value);

        } catch (JMSException e) {

            logger.error("",e);

        }

    }

}

码农公寓

相关文章