专栏首页JavaEEActiveMQ --- 整合篇

ActiveMQ --- 整合篇

之前说到了activeMQ的一些基本用法,本文将介绍activeMQ如何与spring以及spring boot整合。

一、activeMQ与spring整合

1、需要的依赖:

<dependency>
     <groupId>com.fasterxml.jackson.core</groupId>
     <artifactId>jackson-databind</artifactId>
     <version>2.9.9.3</version>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-jms</artifactId>
    <version>4.3.23.RELEASE</version>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
    <version>5.15.9</version>
</dependency>

2、applicationContext.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: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">
    <!-- 1、开启自动扫描 -->
    <context:component-scan base-package="com.zhu.study"/>
    <!-- 2、配置连接 -->
    <bean id="jmsFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
        <property name="connectionFactory">
            <bean class="org.apache.activemq.ActiveMQConnectionFactory">
                <property name="brokerURL" value="tcp://192.168.x.xx:61616"/>
            </bean>
        </property>
        <property name="maxConnections" value="100"/>
    </bean>
    <!-- 3、配置目的地,队列 -->
    <bean id="destinationQueue" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg index="0" value="spring-active-queue"/>
    </bean>
    <!-- 配置目的地,主题 -->
    <bean id="destinationTopic" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg index="0" value="spring-active-topic"/>
    </bean>
    <!-- 4、配置spring提供的jms模板 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="jmsFactory"/>
        <property name="defaultDestination" ref="destinationQueue"/>
        <!--<property name="defaultDestination" ref="destinationTopic"/>-->
        <property name="messageConverter">
            <!-- 做消息类型转换的 -->
            <bean class="org.springframework.jms.support.converter.SimpleMessageConverter"/>
        </property>
    </bean>
</beans>

配置很简单,就是四步:

  • 开启注解扫描;
  • 配置连接;
  • 配置目的地;
  • 配置spring提供的jmsTemplate.

如果想将目的地由queue换成topic,只需要在配置jmsTemplate的时候,将defaultDestination指向你上面配置的topic即可,然后启动时先启动消费者,其他任何地方不用改。

3、生产者:

@Service
public class Produce {

    @Autowired
    private JmsTemplate jmsTemplate;


    public static void main(String[] args){
        // 1、加载配置
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        // 2、获取bean
        Produce produce = context.getBean(Produce.class);
        // 3、调用jmsTemplate发送消息
        /*produce.jmsTemplate.send(new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                TextMessage message = session.createTextMessage("spring整合activeMQ");
                return message;
            }
        });*/

        // lambda方式:编程口诀(拷贝小括号,写死右箭头,落地大括号)
        produce.jmsTemplate.send((Session session) -> {
            TextMessage message = session.createTextMessage("spring整合activeMQ");
            return message;
        });
        System.out.println("activemq send success!");
    }
}

4、消费者:

@Service
public class Consumer {

    @Autowired
    private JmsTemplate jmsTemplate;

    public static void main(String[] args){
        // 1、加载配置
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
        // 2、获取bean
        Consumer consumer = context.getBean(Consumer.class);
        // 3、消费消息
        String result = (String)consumer.jmsTemplate.receiveAndConvert();
        System.out.println("消费者收到消息:" + result);
    }
}

这样就搞定了。不过上面说了,如果把目的地换成topic要先启动消费者。其实还可以配置监听程序,这样就不需要手动启动消费者了,消费者会一直处于待命状态。先写一个监听的类。

5、监听程序:

@Component
public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        if (null != message && message instanceof TextMessage){
            TextMessage textMessage = (TextMessage) message;
            try {
                System.out.println("从activemq收到消息:" + textMessage.getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}

然后在applicationContext.xml中配置:

 <!-- 5、配置监听程序 -->
 <bean id="jmsListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="jmsFactory"/>
        <property name="destination" ref="destinationTopic"/>
        <property name="messageListener" ref="myMessageListener"/>
 </bean>

这样就只需启动生产者即可了,监听程序会自动监听,收到消息就会进行消费。

二、activeMQ与springboot整合

首先新建一个springboot项目,用来编写生产者代码。 1、需要的依赖:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

2、application.yml:

server:
  port: 6666

spring:
  activemq:
    broker-url: tcp://192.168.x.xx:61616
    user: admin
    password: admin

  # false = Queue;  true = Topic
  jms:
    pub-sub-domain: false

#队列名称
myqueue: boot-activemq

上面配置的是队列,要用主题的话,把上面的false改成true。

3、配置类:

@Component
@EnableJms // 这个注解必不可少
public class ConfigBean {

    @Value("${myqueue}")
    private String myQueue;

    // 1、创建队列
    @Bean
    public Queue queue(){
        return new ActiveMQQueue(myQueue);
    }
}

这个配置类主要就是创建了一个队列,队列名从配置文件中读取。

4、生产者: 这里主要有两种生产模式,一种是触发投递,一种是定时投递。触发投递就是程序启动后,满足某个条件才会去调用发送消息的方法;定时投递就是相当于一个定时任务。

@Component
public class Produce {

    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;
    @Autowired
    private Queue queue;
    // 触发投递
    public void produceMsg(){
        String message = "springboot整合activemq成功";
        jmsMessagingTemplate.convertAndSend(queue,message);
        System.out.println("消息触发投递成功");
    }

    // 定时生产,每隔5秒中向MQ发送一次消息(还需在启动类上加上@EnableScheduling注解)
    @Scheduled(fixedDelay = 5000)
    public void produceMsgScheduled(){
        String message = "定时投递消息";
        jmsMessagingTemplate.convertAndSend(queue,message);
        System.out.println("消息定时投递成功");
    }
}

注意定时投递需要在启动类上加@EnableScheduling注解!要测试定时投递,直接运行spring boot的启动类就好了,就可以看到每隔5秒“消息定时投递成功”就会被打印一次。要测试触发投递,就需要我们手动地去调用produceMsg方法,可以写个如下的测试类:

@RunWith(SpringRunner.class)
@SpringBootTest
public class ActivemqspringbootApplicationTests {

    @Autowired
    private Produce produce;
    @Test
    public void contextLoads() {
        produce.produceMsg();
    }
}

运行这个测试类就可以看到会打印出“消息触发投递成功”,然后程序就会停止。

5、消费者: 我们知道消费消息有两种方式,一种是用receive方法,还有就是监听。用receive方法和spring中的一样,这里讲如何配置监听。

@Component
public class Queue_Consumer {

    @JmsListener(destination = "${myqueue}")
    public void recevice(TextMessage message) throws  Exception{
        System.out.println("消费者收到消息:" + message.getText());
    }
}

没错,就是这么简单!在spring中还需要我们自己新建监听类,然后配置到配置文件中,在springboot中,一个注解就搞定了!

6、发布订阅: 上面用的是队列,主题的使用方法如下:

  • 生产者和消费者项目配置文件中的pub-sub-domain的值改为true。
  • 生产者的配置类中,new的不是队列,而是主题了,如下:
@Bean
public Topic topic(){
     return new ActiveMQTopic(myTopic);
}
  • 生产消息的时候注入主题,而非队列,其他无异。

可以看出,springboot整合activemq比spring整个它简单很多!

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • Java常用设计模式

    工厂是干嘛的,就是用来生产的嘛,这里说的工厂也是用来生产的,它是用来生产对象的。也就是说,有些对象我们可以在工厂里面生产,需要用时直接从工厂里面拿出来即可,而不...

    贪挽懒月
  • oracle转postgreSQL修改点

    从2019年开始,就有一个很火热的话题:“去O化”。O就是oracle,也就是将oracle替换成别的数据库。为什么要去O?大致有以下原因:

    贪挽懒月
  • JUC前言:一、volatile关键字与内存可见性二、原子性三、锁分段机制四、闭锁五、创建线程的方式 --- 实现Callable接口六、Lock同步锁七、等待唤醒机制八、ReadWriterLock

    在Java中,线程部分是一个重点,本篇文章说的JUC也是关于线程的。JUC就是java.util .concurrent工具包的简称。这是一个处理线程的工具包,...

    贪挽懒月
  • 源码分析 | 手写mybait-spring核心功能(干货好文一次学会工厂bean、类代理、bean注册的使用)

    一个知识点的学习过程基本分为;运行helloworld、熟练使用api、源码分析、核心专家。在分析mybaits以及mybatis-spring源码之前,我也只...

    小傅哥
  • 抽象工厂(AbstractFactory)模式-创建型模式

    1、new 的问题 常见的对象创建方法: //创建一个Road对象 Road road=new Road(); new的问题:实现依赖,不能应对具体实例的变化 ...

    郑小超.
  • Java 多态

    由于A这个类没有与B建立联系,会报 ClassCastException ,可以通过类的转型来建立联系

    用户2965768
  • SpringBoot的学习使用

    Spring是一个开源框架,2003 年兴起的一个轻量级的Java 开发框架,作者:Rod Johnson 。

    silentcow
  • SpringBoot入门建站全系列(三十)Mybatis多数据源进行数据库操作

    这就牵扯到分布式事务控制,本篇只是讲述多数据源的使用,下一篇再讲述如何使用分布式事务控制中间件。

    品茗IT
  • java:使用匿名类直接new接口

    java中的匿名类有一个倍儿神奇的用法,见下面代码示例: 1 package contract; 2 3 public interface ISay { 4 ...

    菩提树下的杨过
  • SpringBoot系列教程web篇之全局异常处理

    当我们的后端应用出现异常时,通常会将异常状况包装之后再返回给调用方或者前端,在实际的项目中,不可能对每一个地方都做好异常处理,再优雅的代码也可能抛出异常,那么在...

    一灰灰blog

扫码关注云+社区

领取腾讯云代金券