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>.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>.0.7</version>
    </dependency>

2. 配置文件

spring:activemq:
user: admin
password: admin
broker-url: failover:(tcp://.168.43.666:61616)
#是否信任所有包(如果传递的是对象则需要设置为true，默认是传字符串)
packages:
  trust-all: true
#连接池
pool:
  enabled: true
  max-connections:
  idle-timeout:
#  expiry-timeout:
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=-ms
  # 是否在启动时创建连接。可以在启动时用于加热池。
  #spring.activemq.pool.create-connection-on-startup=true
  # 是否用Pooledconnectionfactory代替普通的ConnectionFactory。
  #spring.activemq.pool.enabled=false
  # 连接过期超时。
  #spring.activemq.pool.expiry-timeout=ms
  # 连接空闲超时
  #spring.activemq.pool.idle-timeout=s
  # 连接池最大连接数
  #spring.activemq.pool.max-connections=
  # 每个连接的有效会话的最大数目。
  #spring.activemq.pool.maximum-active-session-per-connection=
  # 当有"JMSException"时尝试重新连接
  #spring.activemq.pool.reconnect-on-exception=true
  # 在空闲连接清除线程之间运行的时间。当为负数时，没有空闲连接驱逐线程运行。
  #spring.activemq.pool.time-between-expiration-check=-ms
  # 是否只使用一个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-11-02
 */
@RestController
@RequestMapping(value = "/active")
public class SendController {  //也可以注入JmsTemplate，JmsMessagingTemplate对JmsTemplate进行了封装
@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;
  /**
 * 发送消息接口
 * 发送queue消息 ：http://.0.0.1:8080/active/send?msg=ceshi1234
 * 发送topic 消息： http://.0.0.1:8080/active/topic/send?msg=ceshi1234
 * 发送queue消息(延迟time毫秒) ：http://.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 >) {
        //延迟队列，延迟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-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 = ：自动确认
     * CLIENT_ACKNOWLEDGE =：客户端手动确认
     * DUPS_OK_ACKNOWLEDGE =： 自动批量确认
     * SESSION_TRANSACTED =：事务提交并确认
     * 但是在activemq补充了一个自定义的ACK模式:
     * INDIVIDUAL_ACKNOWLEDGE =：单条消息确认
     **/
    bean.setSessionAcknowledgeMode();
    //此处设置消息重发规则，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();
    //设置为发布订阅方式, 默认情况下使用的生产消费者方式
    bean.setPubSubDomain(true);
    bean.setConnectionFactory(connectionFactory);
    return bean;
}
  /**
 * 消息的重发规则配置
 */
@Bean
public RedeliveryPolicy redeliveryPolicy() {
    RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
    // 是否在每次尝试重新发送失败后,增长这个等待时间
    redeliveryPolicy.setUseExponentialBackOff(true);
    // 重发次数五次， 总共六次
    redeliveryPolicy.setMaximumRedeliveries();
    // 重发时间间隔,默认为ms（1秒）
    redeliveryPolicy.setInitialRedeliveryDelay();
    // 重发时长递增的时间倍数
    redeliveryPolicy.setBackOffMultiplier();
    // 是否避免消息碰撞
    redeliveryPolicy.setUseCollisionAvoidance(false);
    // 设置重发最大拖延时间-表示无延迟限制
    redeliveryPolicy.setMaximumRedeliveryDelay(-);
    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-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-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 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();
    }
}
}

6. ActiveMQ 消息存储规则

QUEUE 点对点：

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

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

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

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

举例：

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

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

正常新增的只会消费新的topic消息。