天天看点
返回

53分布式电商项目 - Spring集成ActiveMQ

代码已上传到Gtihub

地址：https://github.com/ylw-github/Spring-ActiveMQ-Demo.git

下面讲解Spring整合ActiveMQ过程，根据ActiveMQ的消息通讯机制，分点对点的方式和发布/订阅的模式讲解。

点对点模式

1.消息生产者

（1）创建工程 springjms_producer，在 POM 文件中引入 SpringJms 、activeMQ 以及单元测

试相关依赖

（2）在 src/main/resources 下创建 spring 配置文件 applicationContext-jms-producer.xml

<?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:p="http://www.springframework.org/schema/p"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

    <context:component-scan
        base-package="com.pyg">
    </context:component-scan>

    <!-- 真正可以产生 Connection 的 ConnectionFactory，由对应的 JMS 服务厂商提供-->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </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="queueTextDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="queue_text"/>
    </bean>
</beans>

（3）创建消息生产者类

package com.pyg;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

@Component
public class QueueProducer {

    @Autowired
    private JmsTemplate jmsTemplate;

    @Autowired
    private Destination queueTextDestination;

    /**
     * 发送文本消息
     *
     * @param text
     */
    public void sendTextMessage(final String text) {
        jmsTemplate.send(queueTextDestination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(text);
            }
        });
    }
}

（4）单元测试在 src/test/java 创建测试类

import com.pyg.QueueProducer;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext-jms-producer.xml")
public class TestP2P_Producer {
    @Autowired
    private QueueProducer queueProducer;

    @Test
    public void testSend() {
        queueProducer.sendTextMessage("SpringJms-点对点");
    }
}

2.消息消费者

（1）创建工程 springjms_consumer，在 POM 文件中引入依赖 （同上一个工程）

（2）创建配置文件 applicationContext-jms-consumer-queue.xml

<?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:p="http://www.springframework.org/schema/p"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

    <!-- 真正可以产生 Connection 的 ConnectionFactory，由对应的 JMS 服务厂商提供-->
    <bean id="targetConnectionFactory"
          class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </bean>
    <!-- Spring 用于管理真正的 ConnectionFactory 的 ConnectionFactory -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标 ConnectionFactory 对应真实的可以产生 JMS Connection 的 ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>
    <!--这个是队列目的地，点对点的 文本信息-->
    <bean id="queueTextDestination"
          class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="queue_text"/>
    </bean>
    <!-- 我的监听类 -->
    <bean id="myMessageListener" class="com.pyg.MyMessageListener"></bean>
    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="queueTextDestination"/>
        <property name="messageListener" ref="myMessageListener"/>
    </bean>
</beans>

（3）编写监听类

package com.pyg;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

public class MyMessageListener implements MessageListener {
    public void onMessage(Message message) {
        TextMessage textMessage = (TextMessage) message;
        try {
            System.out.println("接收到消息：" + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

（4）创建测试类

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.IOException;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext-jms-consumer-queue.xml")
public class TestP2P_Consumer {
    @Test
    public void testQueue() {
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

发布/订阅模式

（1）在工程 springjms_producer 的 applicationContext-jms-producer.xml 增加配置。

<!--这个是订阅模式 文本信息-->
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="topic_text"/>
    </bean>

（2）创建生产者类

package com.pyg;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Component;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

@Component
public class TopicProducer {
    @Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private Destination topicTextDestination;

    /**
     * 发送文本消息
     *
     * @param text
     */
    public void sendTextMessage(final String text) {
        jmsTemplate.send(topicTextDestination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(text);
            }
        });
    }
}

（3）编写测试类

import com.pyg.TopicProducer;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext-jms-producer.xml")
public class TestTopic_Producer {
    @Autowired
    private TopicProducer topicProducer;

    @Test
    public void sendTextQueue() {
        topicProducer.sendTextMessage("topic send 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:p="http://www.springframework.org/schema/p"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">

    <!-- 真正可以产生 Connection 的 ConnectionFactory，由对应的 JMS 服务厂商提供-->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://192.168.25.128:61616"/>
    </bean>

    <!-- Spring 用于管理真正的 ConnectionFactory 的 ConnectionFactory -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.SingleConnectionFactory">
        <!-- 目标 ConnectionFactory 对应真实的可以产生 JMS Connection 的 ConnectionFactory -->
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

    <!--这个是队列目的地，点对点的 文本信息-->
    <bean id="topicTextDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="topic_text"/>
    </bean>

    <!-- 我的监听类 -->
    <bean id="myMessageListener" class="com.pyg.MyMessageListener"></bean>

    <!-- 消息监听容器 -->
    <bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="topicTextDestination"/>
        <property name="messageListener" ref="myMessageListener"/>
    </bean>
</beans>
      
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.IOException;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:applicationContext-jms-consumer-topic.xml")
public class TestTopic_Consumer {
    @Test
    public void testQueue() {
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
# 分布式电商
上一篇: 18分布式电商项目 - Spring Security框架
下一篇: 122分布式电商项目 - 终章

继续阅读