SpringBoot整合ActiveMQ的详细步骤

更新时间：2022年11月03日 10:08:21 作者：gblfy

昨天仔细研究了activeMQ消息队列,也遇到了些坑,下面这篇文章主要给大家介绍了关于SpringBoot整合ActiveMQ的相关资料,文中通过实例代码介绍的非常详细,需要的朋友可以参考下

1. 引入依赖

pom文件引入activemq依赖

    <!--activeMq配置-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-activemq</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.activemq</groupId>
            <artifactId>activemq-pool</artifactId>
            <version>5.15.3</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.7</version>
        </dependency>

2. 配置文件

spring:
  activemq:
    user: admin
    password: admin
    broker-url: failover:(tcp://192.168.43.666:61616)
    #是否信任所有包(如果传递的是对象则需要设置为true，默认是传字符串)
    packages:
      trust-all: true
    #连接池
    pool:
      enabled: true
      max-connections: 5
      idle-timeout: 30000
#      expiry-timeout: 0
    jms:
      #默认使用queue模式，使用topic则需要设置为true
      pub-sub-domain: true

      # 是否信任所有包
      #spring.activemq.packages.trust-all=
      # 要信任的特定包的逗号分隔列表（当不信任所有包时）
      #spring.activemq.packages.trusted=
      # 当连接请求和池满时是否阻塞。设置false会抛“JMSException异常”。
      #spring.activemq.pool.block-if-full=true
      # 如果池仍然满，则在抛出异常前阻塞时间。
      #spring.activemq.pool.block-if-full-timeout=-1ms
      # 是否在启动时创建连接。可以在启动时用于加热池。
      #spring.activemq.pool.create-connection-on-startup=true
      # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
      #spring.activemq.pool.enabled=false
      # 连接过期超时。
      #spring.activemq.pool.expiry-timeout=0ms
      # 连接空闲超时
      #spring.activemq.pool.idle-timeout=30s
      # 连接池最大连接数
      #spring.activemq.pool.max-connections=1
      # 每个连接的有效会话的最大数目。
      #spring.activemq.pool.maximum-active-session-per-connection=500
      # 当有"JMSException"时尝试重新连接
      #spring.activemq.pool.reconnect-on-exception=true
      # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
      #spring.activemq.pool.time-between-expiration-check=-1ms
      # 是否只使用一个MessageProducer
      #spring.activemq.pool.use-anonymous-producers=true

3. 生产者

package com.gblfy.producer;

import org.apache.activemq.ScheduledMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.jms.JmsProperties;
import org.springframework.jms.core.JmsMessagingTemplate;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.jms.*;
import java.io.Serializable;

/**
 * 发送消息
 *
 * @author gblfy
 * @date 2022-11-02
 */
@RestController
@RequestMapping(value = "/active")
public class SendController {
    //也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
    @Autowired
    private JmsMessagingTemplate jmsMessagingTemplate;

    /**
     * 发送消息接口
     * 发送queue消息 ：http://127.0.0.1:8080/active/send?msg=ceshi1234
     * 发送topic 消息： http://127.0.0.1:8080/active/topic/send?msg=ceshi1234
     * 发送queue消息(延迟time毫秒) ：http://127.0.0.1:8080/active/send?msg=ceshi1234&time=5000
     *
     * @param msg  消息
     * @param type url中参数,非必须
     * @param time
     * @return
     */
    @RequestMapping({"/send", "/{type}/send"})
    public String send(@PathVariable(value = "type", required = false) String type, String msg, Long time) {
        Destination destination = null;
        if (type == null) {
            type = "";
        }
        switch (type) {
            case "topic":
                //发送广播消息
                destination = new ActiveMQTopic("active.topic");
                break;
            default:
                //发送 队列消息
                destination = new ActiveMQQueue("active.queue");
                break;
        }
        // System.out.println("开始请求发送："+DateUtil.getStringDate(new Date(),"yyyy-MM-dd HH:mm:ss"));
        if (time != null && time > 0) {
            //延迟队列，延迟time毫秒
            //延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true"
            delaySend(destination, msg, time);
        } else {
            jmsMessagingTemplate.convertAndSend(destination, msg);//无序
            //jmsMessagingTemplate.convertSendAndReceive();//有序
        }
        return "activemq消息发送成功 队列消息：" + msg;
    }

    /**
     * 延时发送
     * 说明：延迟队列需要在 <broker>标签上增加属性 schedulerSupport="true"
     *
     * @param destination 发送的队列
     * @param data        发送的消息
     * @param time        延迟时间 /毫秒
     */
    public <T extends Serializable> void delaySend(Destination destination, T data, Long time) {
        Connection connection = null;
        Session session = null;
        MessageProducer producer = null;
        // 获取连接工厂
        ConnectionFactory connectionFactory = jmsMessagingTemplate.getConnectionFactory();
        try {
            // 获取连接
            connection = connectionFactory.createConnection();
            connection.start();
            // 获取session，true开启事务，false关闭事务
            session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
            // 创建一个消息队列
            producer = session.createProducer(destination);
            producer.setDeliveryMode(JmsProperties.DeliveryMode.PERSISTENT.getValue());
            ObjectMessage message = session.createObjectMessage(data);
            //设置延迟时间
            message.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, time);
            // 发送消息
            producer.send(message);
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (producer != null) {
                    producer.close();
                }
                if (session != null) {
                    session.close();
                }
                if (connection != null) {
                    connection.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

4. 配置config

package com.gblfy.config;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.beans.factory.annotation.Value;
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.Queue;
import javax.jms.Topic;

/**
 * 描述:
 * activemq 有两种模式 queue 和 topic
 * queue 模式是单对单，有多个消费者的情况下则是使用轮询监听
 * topic 模式/广播模式/发布订阅模式 是一对多，发送消息所有的消费者都能够监听到
 *
 * @author gblfy
 * @date 2022-11-02
 */
@EnableJms
@Configuration
public class ActiveMQConfig {
    //队列名
    private static final String queueName = "active.queue";
    //主题名
    private static final String topicName = "active.topic";

    @Value("${spring.activemq.user:}")
    private String username;
    @Value("${spring.activemq.password:}")
    private String password;
    @Value("${spring.activemq.broker-url:}")
    private String brokerUrl;

    @Bean
    public Queue acQueue() {
        return new ActiveMQQueue(queueName);
    }

    @Bean
    public Topic acTopic() {
        return new ActiveMQTopic(topicName);
    }

    @Bean
    public ActiveMQConnectionFactory connectionFactory() {
        return new ActiveMQConnectionFactory(username, password, brokerUrl);
    }

    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        // 关闭Session事务，手动确认与事务冲突
        bean.setSessionTransacted(false);
        // 设置消息的签收模式（自己签收）
        /**
         * AUTO_ACKNOWLEDGE = 1 ：自动确认
         * CLIENT_ACKNOWLEDGE = 2：客户端手动确认
         * DUPS_OK_ACKNOWLEDGE = 3： 自动批量确认
         * SESSION_TRANSACTED = 0：事务提交并确认
         * 但是在activemq补充了一个自定义的ACK模式:
         * INDIVIDUAL_ACKNOWLEDGE = 4：单条消息确认
         **/
        bean.setSessionAcknowledgeMode(4);
        //此处设置消息重发规则，redeliveryPolicy() 中定义
        connectionFactory.setRedeliveryPolicy(redeliveryPolicy());
        bean.setConnectionFactory(connectionFactory);
        return bean;
    }

    @Bean
    public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        // 关闭Session事务，手动确认与事务冲突
        bean.setSessionTransacted(false);
        bean.setSessionAcknowledgeMode(4);
        //设置为发布订阅方式, 默认情况下使用的生产消费者方式
        bean.setPubSubDomain(true);
        bean.setConnectionFactory(connectionFactory);
        return bean;
    }

    /**
     * 消息的重发规则配置
     */
    @Bean
    public RedeliveryPolicy redeliveryPolicy() {
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        // 是否在每次尝试重新发送失败后,增长这个等待时间
        redeliveryPolicy.setUseExponentialBackOff(true);
        // 重发次数五次， 总共六次
        redeliveryPolicy.setMaximumRedeliveries(5);
        // 重发时间间隔,默认为1000ms（1秒）
        redeliveryPolicy.setInitialRedeliveryDelay(1000);
        // 重发时长递增的时间倍数2
        redeliveryPolicy.setBackOffMultiplier(2);
        // 是否避免消息碰撞
        redeliveryPolicy.setUseCollisionAvoidance(false);
        // 设置重发最大拖延时间-1表示无延迟限制
        redeliveryPolicy.setMaximumRedeliveryDelay(-1);
        return redeliveryPolicy;
    }
}

5. queue消费者

package com.gblfy.listener;

import org.apache.activemq.command.ActiveMQMessage;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Session;

/**
 * TODO
 *
 * @author gblfy
 * @Date 2022-11-02
 **/
@Component
public class QueueListener {

    /**
     * queue 模式 单对单，两个消费者监听同一个队列则通过轮询接收消息
     * containerFactory属性的值关联config类中的声明
     *
     * @param msg
     */
    @JmsListener(destination = "active.queue", containerFactory = "jmsListenerContainerQueue")
    public void queueListener(ActiveMQMessage message, Session session, String msg) throws JMSException {
        try {
            System.out.println("active queue 接收到消息 " + msg);
            //手动签收
            message.acknowledge();
        } catch (Exception e) {
            //重新发送
            session.recover();
        }
    }
}

6. topic消费者

package com.gblfy.listener;

import org.apache.activemq.command.ActiveMQMessage;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import javax.jms.JMSException;
import javax.jms.Session;

/**
 * TODO
 *
 * @author gblfy
 * @Date 2022-11-02
 **/
@Component
public class TopicListener {

    /**
     * topic 模式/广播模式/发布订阅模式 一对多，多个消费者可同时接收到消息
     * topic 模式无死信队列，死信队列是queue模式
     * containerFactory属性的值关联config类中的声明
     *
     * @param msg
     */
    @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic")
    public void topicListener(ActiveMQMessage message, Session session, String msg) throws JMSException {
        try {
            // System.out.println("接收到消息：" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss"));
            System.out.println("active topic 接收到消息 " + msg);
            System.out.println("");
            //手动签收
            message.acknowledge();
        } catch (Exception e) {
            //重新发送
            session.recover();
        }
    }

    @JmsListener(destination = "active.topic", containerFactory = "jmsListenerContainerTopic")
    public void topicListener2(ActiveMQMessage message, Session session, String msg) throws JMSException {
        try {
            // System.out.println("接收到消息：" + DateUtil.getStringDate(new Date(), "yyyy-MM-dd HH:mm:ss"));
            System.out.println("active topic2 接收到消息 " + msg);
            System.out.println("");
            //手动签收
            message.acknowledge();
        } catch (Exception e) {
            //重新发送
            session.recover();
        }
    }
}

6. ActiveMQ 消息存储规则

QUEUE 点对点：

特点：消息遵循先到先得，消息只能被一个消费者消费。

消息存储规则：消费者消费消息成功，MQ服务端消息删除

TOPIC订阅模式: 消息属于广播（订阅）模式，消息会被所有的topic消费者消费消息。

消息存储规则：所有消费者消费成功，MQ服务端消息删除，有一个消息没有没有消费完成，消息也会存储在MQ服务端。

举例：

已经处于运行topic消费者5个，5个消费者消费完成后，MQ服务端消息删除。

扩展点补充：如果想额外添加topic消费者，如果MQ服务端消息没有被消费完毕，新增topic消费者可以消费以前未被消费的消息，
正常新增的只会消费新的topic消息。

总结

到此这篇关于SpringBoot整合ActiveMQ的文章就介绍到这了,更多相关SpringBoot整合ActiveMQ内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Mac电脑安装多个JDK版本的详细图文教程
目前使用的主流版本还是JDK 8,但偶尔会想体验下新版本(或者旧版本),如果能装多个版本的JDK,而且很方便的切换就好了,这篇文章主要给大家介绍了关于Mac电脑安装多个JDK版本的相关资料,需要的朋友可以参考下
2024-03-03
SpringBoot整合ActiveMQ过程解析
这篇文章主要介绍了SpringBoot整合ActiveMQ过程解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-09-09
SpringBoot简单实现文件上传
这篇文章主要介绍了SpringBoot简单实现文件上传，文章围绕主题展开详细的内容介绍，具有一定的参考价值，感兴趣的小伙伴可以参考一下
2022-09-09
Spring 异步接口返回结果的四种方式
这篇文章主要介绍了Spring 异步接口返回结果的四种方式，文章围绕主题展开详细的内容介绍，具有一定的参考价值，感兴趣的小伙伴可以参考一下
2022-08-08
解决pageHelper分页失效以及如何配置问题
这篇文章主要介绍了解决pageHelper分页失效以及如何配置问题，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2023-04-04
java中实现四则运算代码
本文给大家分享了几个java中实现四则运算的代码，有个人的也有网友的，代码写的不是很好，难免会有BUG，忘发现BUG的亲们能提醒我下，好让我改进
2015-08-08
Java日期接收报错:could not be parsed, unparsed text found a
在做Java开发时肯定会碰到传递时间参数的情况,这篇文章主要给大家介绍了关于Java日期接收报错:could not be parsed, unparsed text found at index 10的解决办法,文中通过代码介绍的非常详细,需要的朋友可以参考下
2024-01-01
Java中Runnable和Thread的区别分析
在java中可有两种方式实现多线程，一种是继承Thread类，一种是实现Runnable接口,下面就拉分别介绍一下这两种方法的优缺点
2013-03-03
Java并发之异步的八种实现方式
本文主要介绍了Java并发之异步的八种实现方式，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-06-06
Java+ElasticSearch+Pytorch实现以图搜图功能
这篇文章主要为大家详细介绍了Java如何利用ElasticSearch和Pytorch实现以图搜图功能，文中的示例代码讲解详细，具有一定的学习价值，感兴趣的小伙伴可以了解一下
2023-06-06