Spring boot 集成 ActiveMQ

Donne 发布于2019-08-19 11:43 / 2779人阅读

摘要：安装到官方网站下载最新的的安装包，并解压到本地目录下，下载链接如下。修改消费者使用配置消费者监听的队列，其中是接收到的消息收到的报文为接收到的消息重新执行

安装ActiveMQ

到Apache官方网站下载最新的ActiveMQ的安装包，并解压到本地目录下，下载链接如下：http://activemq.apache.org/do...。

进入bin 目录，如果我们是32位的机器，就双击 win32 目录下的 activemq.bat，如果是64位机器，则双击 win64 目录下的 activemq.bat ，运行结果如下：
成功之后在浏览器输入 http://127.0.0.1:8161/ 地址，可以看到 ActiveMQ 的管理页面，用户名和密码默认都是 admin

Spring Boot 整合 ActiveMQ

工程结构

添加 pom 依赖


    org.springframework.boot
    spring-boot-starter-activemq

config 配置

# activemq
spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
#默认为true表示使用内存的activeMQ，不需要安装activeMQ server
spring.activemq.in-memory=true
 
#如果此处设置为true，需要加如下的依赖包
#          org.apache.activemq
#          activemq-pool
# 否则会自动配置失败，报JmsMessagingTemplate注入失败
spring.activemq.pool.enabled=false

队列模式

创建消息提供者：Producer.java

import javax.jms.Destination;
 
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.stereotype.Service;
 
@Service
public class Producer {
 
    @Autowired // 也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
    private JmsMessagingTemplate jmsTemplate;
 
    // 发送消息，destination是发送到的队列，message是待发送的消息
    public void sendMessage(Destination destination, final String message){
        jmsTemplate.convertAndSend(destination, message);
    }
}

创建消费者一: Consumer.java

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
 
@Component
public class Consumer {
 
    // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue")
    public void receiveQueue(String text) {
        System.out.println("Consumer收到的报文为:"+text);
    }
}

创建消费者二：Consumer1 .java

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
 
@Component
public class Consumer1 {
 
    // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue")
    public void receiveQueue(String text) {
 
        System.out.println("Consumer1收到的报文为:"+text);
    }
}

测试

创建一个 ActivceMQQueue 对象，表示队列模式，下面会介绍主题模式。

import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
import javax.jms.Destination;
 
@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsApplicationTests {
 
    @Autowired
    private Producer producer;
 
    @Test
    public void contextLoads() {
 
        Destination destination = new ActiveMQQueue("mytest.queue");
 
        for(int i=0; i<100; i++){
            producer.sendMessage(destination, "myname is chhliu!!!");
        }
    }
}

双向队列

使用 @SendTo 注解可以将方法的返回值重新放入到消息队列中，供其他消费者消费

这里我们修改 Consumer1.java 增加注解 @SendTo

public class Consumer1 {
 
    // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue")
    @SendTo("out.queue")
    public String receiveQueue(String text) {
 
        System.out.println("Consumer1收到的报文为:"+text);
 
        return "return message" + text;
    }
 
    @JmsListener(destination = "out.queue")
    public void outQueue(String text){
        System.out.println("Consumer1 outQueue:"+text);
    }
}

主题模式

配置文件中需要增加

# 启用 topic 模式
spring.jms.pub-sub-domain=true

修改 Consumer1 增加对主题的监听

import org.springframework.jms.annotation.JmsListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
 
@Component
public class Consumer1 {
 
    // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue")
    @SendTo("out.queue")
    public String receiveQueue(String text) {
 
        System.out.println("Consumer1 收到的报文为:"+text);
 
        return "return message" + text;
    }
 
    @JmsListener(destination = "out.queue")
    public void outQueue(String text){
        System.out.println("Consumer1 outQueue:"+text);
    }
 
    @JmsListener(destination = "mytest.topic")
    public void topicQueue(String text){
        System.out.println("Consumer 接收到的 topic 消息：" + text);
    }
}

修改测试类

增加发送主题消息

import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
 
import javax.jms.Destination;
 
@RunWith(SpringRunner.class)
@SpringBootTest
public class JmsApplicationTests {
 
    @Autowired
    private Producer producer;
 
    @Test
    public void contextLoads() {
 
        Destination destination = new ActiveMQQueue("mytest.queue");
        Destination topicDestination = new ActiveMQTopic("mytest.topic");
 
        for(int i=0; i<10; i++){
            producer.sendMessage(destination, "Queue Message......");
            producer.sendMessage(topicDestination, "Topic Message!!!");
        }
    }
}

消息的结果为：

没有出现队列消息。此时需要修改我们的监听，指定出是哪种类型的

增加配置类

这个配置类没有在那个工程图上面出现，添加一个类就好。

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.annotation.EnableJms;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.config.JmsListenerContainerFactory;
 
import javax.jms.ConnectionFactory;
 
@Configuration
@EnableJms
public class JmsConfig {
    @Bean
    public JmsListenerContainerFactory topicListenerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(true);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
 
    @Bean
    public JmsListenerContainerFactory queueListenerFactory(ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setPubSubDomain(false);
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
 
}

修改消费者 Consumer1.java

import org.springframework.jms.annotation.JmsListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;
 
@Component
public class Consumer1 {
 
    // 使用JmsListener配置消费者监听的队列，其中text是接收到的消息
    @JmsListener(destination = "mytest.queue", containerFactory = "queueListenerFactory")
    @SendTo("out.queue")
    public String receiveQueue(String text) {
 
        System.out.println("Consumer1 收到的报文为:"+text);
 
        return "return message" + text;
    }
 
    @JmsListener(destination = "out.queue", containerFactory = "queueListenerFactory")
    public void outQueue(String text){
        System.out.println("Consumer1 outQueue:"+text);
    }
 
    @JmsListener(destination = "mytest.topic", containerFactory = "topicListenerFactory")
    public void topicQueue(String text){
        System.out.println("Consumer 接收到的 topic 消息：" + text);
    }
}

重新执行

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/77730.html

Spring Boot 参考指南（消息传递）

摘要：还自动配置发送和接收消息所需的基础设施。支持是一个轻量级的可靠的可伸缩的可移植的消息代理，基于协议，使用通过协议进行通信。 32. 消息传递 Spring框架为与消息传递系统集成提供了广泛的支持，从使用JmsTemplate简化的JMS API到使用完整的基础设施异步接收消息，Spring AMQP为高级消息队列协议提供了类似的特性集。Spring Boot还为RabbitTempla...

Doyle 2019-08-16 12:31 评论0 收藏0
慕课网_《Java消息中间件》学习总结

摘要：时间年月日星期六说明本文部分内容均来自慕课网。这个时候，可以启动多台积分系统，来同时消费这个消息中间件里面的登录消息，达到横向扩展的作用。时间：2017年07月22日星期六说明：本文部分内容均来自慕课网。@慕课网：http://www.imooc.com教学源码：无学习源码：https://github.com/zccodere/s... 第一章：课程介绍 1-1 课程安排 Java...

twohappy 2019-08-15 11:00 评论0 收藏0
SpringBoot ActiveMQ 整合使用

摘要：介绍它是出品，最流行的，能力强劲的开源消息总线。是一个完全支持和规范的实现，尽管规范出台已经是很久的事情了，但是在当今的应用中间仍然扮演着特殊的地位。相关文章整合使用整合使用关注我转载请务必注明原创地址为安装同之前一样，直接在里面玩吧。 showImg(https://segmentfault.com/img/remote/1460000012996066?w=1920&h=1281)...

gaara 2019-08-15 13:54 评论0 收藏0
JMS 在 SpringBoot 中的使用

摘要：本文主要讲述消息服务在中的使用。所以需要一个监听容器工厂的概念，即接口，它会引用上面创建好的与的连接工厂，由它来负责接收消息以及将消息分发给指定的监听器。为了消费消息，订阅者必须保持运行的状态。 JMS 在 SpringBoot 中的使用摘要：本文属于原创，欢迎转载，转载请保留出处：https://github.com/jasonGeng88/blog> 本文所有服务均采用doc...

Michael_Ding 2019-08-14 18:43 评论0 收藏0
使用Spring/Spring Boot集成JMS的陷阱

摘要：本文旨在指出中集成的一些性能陷阱，在另一篇文章各组件详解里有组件介绍及如何正确使用的内容。因此的做法会大大降低性能，并且将大部分的时间都花在反复重建这些对象上。提供的可以让使用避免频繁创建的问题。至于使用的性能测试则留给同学自己做了。 Github 本文旨在指出Spring/Spring Boot中集成JMS的一些性能陷阱，在另一篇文章Spring JMS各组件详解里有Spring J...

xcold 2019-08-14 14:18 评论0 收藏0