`

Sping整合ActiveMQ（四.实例详解）

JavaJmsSpringActiveMQ 
阅读更多

一.Pom文件

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

  <modelVersion>4.0.0</modelVersion>

  <groupId>ActiveMQ-Spring</groupId>
  <artifactId>ActiveMQ-Spring</artifactId>
  <packaging>war</packaging>
  <version>1.0-SNAPSHOT</version>

  <name>ActiveMQ-Spring4 Maven Webapp</name>
  <url>http://maven.apache.org</url>

    <properties>
        <spring-version>3.2.6.RELEASE</spring-version>
        <activemq-version>5.6.0</activemq-version>
        <jms-version>2.0</jms-version>
    </properties>

    <dependencies>
        <!--Junit-->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.10</version>
        </dependency>
        <!--Slf4j-->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.5</version>
        </dependency>
        <!--Spring-->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-core</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context-support</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-tx</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-web</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <!--Spring.Jms-->
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-jms</artifactId>
            <version>${spring-version}</version>
        </dependency>
        <!--Javax.Jms-->
        <dependency>
            <groupId>javax.jms</groupId>
            <artifactId>javax.jms-api</artifactId>
            <version>${jms-version}</version>
        </dependency>
        <!--ActiveMQ-->
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-all</artifactId>
            <version>${activemq-version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>${activemq-version}</version>
        </dependency>
        <!--xbean-spring-->
        <dependency>
            <groupId>org.apache.xbean</groupId>
            <artifactId>xbean-spring</artifactId>
            <version>3.16</version>
        </dependency>

    </dependencies>

  <build>
    <finalName>ActiveMQ-Spring</finalName>
  </build>

</project>

 

二.Spring配置文件

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

    <context:annotation-config/>
    <context:component-scan base-package="com.guangjieba.whale.reptile.**"/>

    <!--配置ActionMQ的连接工厂,用amq配置-->
    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
    </bean>
    <!--<amq:connectionFactory id="connectionFactory" brokerURL="tcp://localhost:61616"/>-->

    <!--配置Queue消息目的地,用amq配置-->
    <bean id="quartzQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="QuartzQueue"/>
        <!--<constructor-arg index="0" value="QuartzQueue"/>-->  <!--带index参数-->
    </bean>
    <bean id="realTimeQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="RealTimeQueue"/>
    </bean>
    <!--<amq:queue id="queue" physicalName="FirstQueue"/>-->

    <!--配置Topic消息目的地-->
    <!--<bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg index="0" value="MyTopic"/>
    </bean>-->

    <!--配置JmsTemplate：手动，有超时时间-->
    <bean id="jmsTemplate1" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="defaultDestination" ref="quartzQueue"/>
        <property name="receiveTimeout" value="2000" />
    </bean>
    <!--配置JmsTemplate：即时——通过监听器实现-->
    <bean id="jmsTemplate2" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="realTimeQueue" />
    </bean>

    <!--配置消息监听（方式一），消息入队通知接收器，即时出队-->
    <bean id="listenerContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="concurrentConsumers" value="10" />
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destinationName" value="RealTimeQueue" />    
        <property name="messageListener" ref="realTimeActiveMQ" />   <!--被监听的类-->
        <property name="pubSubNoLocal" value="false"/>
    </bean>

    <!--配置消息监听（方式二），消息入队通知接收器，即时出队-->
    <jms:listener-container connection-factory="connectionFactory">
        <jms:listener destination="RealTimeQueue" ref="realTimeActiveMQ"/>
        <!--<jms:listener destination="RealTimeQueue" ref="realTimeActiveMQ" method="receive"/>-->
    </jms:listener-container>
</beans>
 

 三.ActiveMQ类1：手动型

import com.huaxia.entity.Goods
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.*;

/**
 * Created with IntelliJ IDEA..
 * User: Leon
 * Date: 14-2-28
 * Time: 下午4:52
 * To change this template use File | Settings | File Templates.
 */
@Service
public class QuartzActiveMQ {

    /*注：因为配置文件中有两个JmsTemplate,故用@Resource,可byName注入*/
    @Resource
    private JmsTemplate jmsTemplate1;

    /**
     * 直接将消息转化并入队
     * @param rbGoods 发送的商品对象
     */
    public void simpleSend(final Goods goods) {
        jmsTemplate1.convertAndSend(goods);
    }

    /**
     * 发送消息到队列中
     * @param rbGoods 发送的商品对象
     */
    public void send(final Goods goods) {
        jmsTemplate1.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createObjectMessage(goods);
            }
        });
    }

    /**
     * 从队列中接收消息并返回（手动方式）
     */
    public Goods receive() {
        ObjectMessage message = (ObjectMessage) jmsTemplate1.receive();
        if (message == null) {
            return null;
        }
        try {
            Goods goods = (Goods) message.getObject();
            return goods;
        } catch (JMSException e) {
            e.printStackTrace();
        }
        return null;
    }

 

 四.ActiveMQ类2：即时通信型

import com.huaxia.entity.Goods;
import com.huaxia.dao.GoodsRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

/**
 * Created with IntelliJ IDEA..
 * User: Leon
 * Date: 14-3-1
 * Time: 上午10:10
 * To change this template use File | Settings | File Templates.
 * 通过监听器实时监听并接收消息
 */
@Service
public class RealTimeActiveMQ implements MessageListener {

    @Resource
    private JmsTemplate jmsTemplate2;
    @Autowired
    private GoodsRepository roodsRepository;

    /**
     * 商品入队
     * @param rbGoods 商品
     */
    public void send(final Goods goods){
        jmsTemplate2.convertAndSend(goods);
    }

    /**
     * 商品即时出队并持久化
     * @param message
     */
    @Override
    public void onMessage(Message message) {
        if (message instanceof ObjectMessage) {
            try {
                Goods goods = (Goods) ((ObjectMessage) message).getObject();
                Goods roodsTemp = goodsRepository.save(goods);
                if (goodsTemp.getId() == null) {
                    send(goods);
                }
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

 

 五.接口

 

@Controller
@RequestMapping("/brandsource")
public class BrandSourceController {

    @Autowired
    private QuartzActiveMQ activeMQController;、
    @Autowired
    private IGoodsRestTemplate iGoodsRestTemplate;
 
    @RequestMapping(value = "/goods/check", method = RequestMethod.GET)
    private void checkGoods() {
        while (true) {
            int count = 0;
            RbGoods rbGoods = activeMQController.receive();
            if (rbGoods != null) {
                boolean flag = iGoodsRestTemplate.postUpdateGoodsStatus(rbGoods);
                while (!flag) {
                    count += 1;
                    if (count == 3) {
                        activeMQController.send(rbGoods);
                        break;
                    }
                    flag = iGoodsRestTemplate.postUpdateGoodsStatus(rbGoods);
                }
                continue;
            }
            else {
                break;
            }
        }
    }

}

 

 

 六.测试类1：手动型

 

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.web.context.WebApplicationContext;

import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.result.MockMvcResultHandlers.print;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;


/**
 * Created with IntelliJ IDEA..
 * User: Leon
 * Date: 14-2-28
 * Time: 下午3:37
 * To change this template use File | Settings | File Templates.
 */
@RunWith(SpringJUnit4ClassRunner.class)
@WebAppConfiguration
@ContextConfiguration({"/ServletContext.xml"})
public class QuartzActiveMQTest {

    @Autowired
    private QuartzActiveMQ activeMQController;
    @Autowired
    private RbGoodsRepository rbGoodsRepository;
    @Autowired
    WebApplicationContext webApplicationContext;
    private MockMvc mockMvc;

    @Before
    public void init() {
        mockMvc = MockMvcBuilders.webAppContextSetup(webApplicationContext).build();
    }

    /**
     * 往队列里添加数据
     */
    @Test
    public void sendTest() {
        RbGoods rbGoods = rbGoodsRepository.findOne(QRbGoods.rbGoods.id.eq("aHR0cDovL3d3dy5obS5jb20vY24vcHJvZHVjdC8wODcwOT9hcnRpY2xlPTA4NzA5LUM"));
        System.out.println(rbGoods);
        /*RbGoods rbGoods = new RbGoods();
        rbGoods.setBrandId("123234");
        rbGoods.setBrandName("sdaf");*/
        activeMQController.send(rbGoods);
    }

    /**
     * 批量审核对接接口的测试
     */
    @Test
    public void checkGoodsTest() {
        try {
            mockMvc.perform(get("/brandsource/goods/check").accept(MediaType.APPLICATION_JSON))
                    .andDo(print())
                    .andExpect(status().isOk())
                    .andReturn();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 从队列里面拿出数据
     */
    @Test
    public void receiveTest() {
        activeMQController.receive();
    }
}

 

 七.测试类2：即时通信型

 

import org.junit.Before;
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;
import org.springframework.test.context.web.WebAppConfiguration;
import org.springframework.test.web.servlet.MockMvc;
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
import org.springframework.web.context.WebApplicationContext;

/**
 * Created with IntelliJ IDEA..
 * User: Leon
 * Date: 14-3-1
 * Time: 上午10:25
 * To change this template use File | Settings | File Templates.
 */
@RunWith(SpringJUnit4ClassRunner.class)
@WebAppConfiguration
@ContextConfiguration({"/ServletContext.xml"})
public class RealTimeActiveMQTest {
    @Autowired
    private RealTimeActiveMQ realTimeActiveMQ;
    @Autowired
    private RbGoodsRepository rbGoodsRepository;
    @Autowired
    WebApplicationContext webApplicationContext;
    private MockMvc mockMvc;

    @Before
    public void init() {
        mockMvc = MockMvcBuilders.webAppContextSetup(webApplicationContext).build();
    }

    @Test
    public void sendTest() {
        realTimeActiveMQ.send(new RbGoods());
    }
}

 

 八.Web.xml配置文件

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
         version="2.5">

    <display-name>Weibo Web Application</display-name>

    <filter>
        <filter-name>CORS Filter</filter-name>
        <filter-class>org.ebaysf.web.cors.CORSFilter</filter-class>
    </filter>
    <filter-mapping>
        <filter-name>CORS Filter</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>

    <filter>
        <filter-name>encodingFilter</filter-name>
        <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
        <init-param>
            <param-name>encoding</param-name>
            <param-value>UTF-8</param-value>
        </init-param>
    </filter>

    <filter-mapping>
        <filter-name>encodingFilter</filter-name>
        <url-pattern>/*</url-pattern>
    </filter-mapping>

    <!-- The definition of the Root Spring Container shared by all Servlets and Filters -->
    <context-param>
        <param-name>contextConfigLocation</param-name>
        <param-value>/WEB-INF/spring/root-context.xml</param-value>
    </context-param>

    <!-- Processes application requests -->
    <servlet>
        <servlet-name>appServlet</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <!--<param-value>/WEB-INF/spring/appServlet/servlet-context.xml</param-value>-->
            <param-value>/WEB-INF/spring/root-context.xml</param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>

    <servlet-mapping>
        <servlet-name>appServlet</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>

    <!-- Creates the Spring Container shared by all Servlets and Filters -->
    <listener>
        <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
    </listener>
    <listener>
        <listener-class>org.springframework.web.context.request.RequestContextListener</listener-class>
    </listener>
    <listener>
        <listener-class>com.guangjieba.whale.reptile.application.InitApp</listener-class>
    </listener>

    <!-- Disables Servlet Container welcome file handling. Needed for compatibility with Servlet 3.0 and Tomcat 7.0 -->
    <welcome-file-list>
        <welcome-file>index.html</welcome-file>
        <welcome-file>index.jsp</welcome-file>
    </welcome-file-list>


</web-app>

 

分享到：
| Sping整合ActiveMQ（三.Spring JMSTemplat ...
评论
1 楼 dongpy1111 2014-04-24  
请问消费者这边必须要用while(true){}这种方式么？这样岂不是很占资源？
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    Spring+ActiveMQ整合实例代码工程

    **Spring与ActiveMQ整合详解** 在Java开发中，Spring框架是极为重要的应用基础，而ActiveMQ作为Apache出品的一款开源消息中间件，常被用于实现应用间的异步通信和解耦。本实例代码工程"Spring+ActiveMQ整合实例代码...

    SpringBoot集成ActiveMQ实例详解.docx

    在本文中，我们将深入探讨如何在Spring Boot应用中集成ActiveMQ，这是一个强大的Java消息服务（JMS）实现。首先，我们需要理解JMS的基本概念。Java消息服务（JMS）是Java平台上的一个标准API，它定义了应用程序如何...

    Spring与ActiveMQ完整实例源码下载 Maven 版.zip

    《Spring与ActiveMQ整合实战详解》 在Java开发领域，Spring框架和ActiveMQ是两个非常重要的组件。Spring作为一款强大的企业级应用框架，提供了一系列的工具和服务，方便开发者构建复杂的应用系统；而ActiveMQ则是...

    activemq-spring-1.2.jar.zip

    《ActiveMQ与Spring整合详解及1.2版本的使用》 在Java消息服务（Java Message Service，JMS）领域，Apache ActiveMQ是一个广泛使用的开源消息代理，它提供了一个高效的、可扩展的消息传递平台。而Spring框架是Java...

    ActiveMq-JMS好用实例详解

    ### ActiveMQ-JMS好用实例详解 #### 一、ActiveMQ简介及特点 **ActiveMQ** 是一个非常流行的开源消息中间件，它基于 **Java消息服务(JMS)** 规范，能够提供高度可靠的消息传递机制。ActiveMQ 以其丰富的功能集、...

    ActiveMQ与Spring整合示例Demo

    **ActiveMQ与Spring整合详解** ActiveMQ是Apache组织开发的一款开源消息中间件，它遵循Java消息服务（JMS）规范，提供可靠的消息传递和高效的消息处理能力。在企业级应用中，ActiveMQ常用于实现应用之间的解耦，...

    ActiveMQ+Spring完整详解例子

    本文将全面讲解ActiveMQ的使用流程，并结合Spring框架进行实例解析，帮助开发者深入理解如何在实际项目中部署和使用ActiveMQ。 1. **ActiveMQ的基本概念** - **消息**: 在ActiveMQ中，消息是数据传输的基本单元，...

    activemq与spring整合

    **ActiveMQ 与 Spring 整合详解** ActiveMQ 和 Spring 的整合是企业级应用中常见的消息中间件解决方案，尤其在构建分布式系统时，两者结合能够提供高效、可靠的异步通信能力。ActiveMQ 是一个开源的消息代理，实现...

    ActiveMQ5.1+Spring2.5 Demo

    **ActiveMQ5.1+Spring2.5 Demo详解** ActiveMQ是Apache软件基金会下的一个开源项目，它是一款功能强大的消息中间件，支持多种消息协议，如AMQP、STOMP、OpenWire等。在版本5.1中，ActiveMQ提供了一流的消息传输性能...

    ActiveMQ与spring集成实例之使用消息监听器

    **ActiveMQ与Spring集成实例详解** ActiveMQ是Apache软件基金会下的一个开源项目，它是一个功能丰富的Java消息服务（JMS）提供商，支持多种协议，并且能够处理大量的并发消息传输。而Spring框架则是一个广泛使用的...

    Spring 实现远程访问详解——jms和activemq

    3. Spring 整合JMS和ActiveMq流程 1) 下载和部署ActiveMq服务器 2) Spring jms和activemq相关依赖引入 3) Spring整合activemq配置 4) 定义消息发布者(生产者) 5) 定义消息订阅者(消费者) 6) Spring mvc配置 7) 实例...

    ActiveMQ快速上手 PDF

    ### ActiveMQ 快速上手知识点详解 #### 一、ActiveMQ简介 - **定义**：ActiveMQ 是 Apache 软件基金会所研发的一款开源消息中间件，它完全支持 JMS 1.1 和 J2EE 1.4 规范，能够作为 JMS Provider 实现消息传递功能...

    精通spring2.x企业应用开发详解

    - **Spring与消息队列集成**：讲解如何利用Spring整合消息中间件，如RabbitMQ或ActiveMQ，实现异步处理和解耦。 - **Spring Batch**：可能介绍Spring的批量处理框架，用于执行大规模数据处理任务。 - **Spring的Web...

    从入门到精通的ActiveMQ.docx

    5. **Spring 整合**：ActiveMQ 可以无缝集成到 Spring 框架中，简化应用程序的开发和配置。 6. **监控与优化**：Web 控制台提供了丰富的监控功能，包括查看消息队列状态、性能指标等，便于系统调优。 通过以上介绍...

    spring-activeMQ-demo:spring-activeMQ-演示

    《Spring与ActiveMQ整合实战详解》 在Java开发领域，消息队列（Message Queue）作为解耦、异步处理和提高系统吞吐量的重要工具，被广泛应用。Spring框架以其强大的集成能力，使得与各种消息中间件如ActiveMQ的整合...

    spring-mq完美整合示例

    《Spring与ActiveMQ整合实践详解》 在现代企业级应用开发中，消息队列（Message Queue，MQ）作为异步处理、解耦系统组件的重要工具，已经广泛应用。本示例聚焦于Spring框架与Apache ActiveMQ的整合，旨在帮助开发者...

    ActiveMQ 与Spring结合示例

    **ActiveMQ 与 Spring 结合使用详解** ActiveMQ 是 Apache 开源组织提供的一个开源消息中间件，它实现了 JMS（Java Message Service）规范，能够帮助应用程序实现异步处理和解耦，提高系统的可扩展性和可靠性。...

    spring与jms结合实例

    ### Spring与JMS结合实例详解 #### 一、引言 在现代企业级应用开发中，消息中间件作为实现系统间解耦、提高系统稳定性的关键组件被广泛应用。Java消息服务（Java Message Service, JMS）是Java平台中关于面向消息...

Magicbox
Global site tag (gtag.js) - Google Analytics