ActiveMQ --- 整合篇

贪挽懒月

发布于 2019-10-22 22:54:44

5400

发布于 2019-10-22 22:54:44

文章被收录于专栏：JavaEEJavaEE

之前说到了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整个它简单很多！

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2019.10.19 ，如有侵权请联系 cloudcommunity@tencent.com 删除

spring

spring boot

xml

腾讯云测试服务

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度