本文共 5243 字，大约阅读时间需要 17 分钟。

Spring Boot 2.x 整合 ActiveMQ 消息中间件服务

项目结构

本项目基于 Spring Boot 2.x 使用 ActiveMQ 作为消息中间件，实现消息的推送和接收功能。本文将展示项目的实现过程，包括依赖管理、配置设置、服务开发和任务调度等内容。

1. 依赖管理

在项目的 pom.xml 中添加相关依赖：

       
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-activemq
         
        
        
            
     
      org.messaginghub
             
     
      pooled-jms
             
     
      1.0.5
         
        
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-web
             
                 
                      
       
        org.springframework.boot
                       
       
        spring-boot-starter-tomcat
                   
              
         
        
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-jetty
         
        
        
            
     
      org.springframework.boot
             
     
      spring-boot-starter-test
             
     
      test
         
    
   

2. 配置文件

在 application.yml 中配置以下内容：

server:  port: 8093spring:  activemq:    broker-url: tcp://192.168.44.113:61616    user: admin    password: admin    pool:      enabled: true      max-connections: 200  jms:    pub-sub-domain: true # true 为 topics，false 为 queues# 服务名称application:  name: boot-activemq

3. ActiveMQ配置类

创建 ActiveMQConfig.java 文件：

package com.hf.boot.config;import org.apache.activemq.command.ActiveMQQueue;import org.apache.activemq.command.ActiveMQTopic;import org.springframework.context.annotation.Bean;import org.springframework.jms.annotation.EnableJms;import org.springframework.stereotype.Component;import javax.jms.Queue;import javax.jms.Topic;@Component@EnableJmspublic class ActiveMQConfig {    @Bean    public Queue queue() {        return new ActiveMQQueue("boot-activemq-queue-01");    }    @Bean    public Topic topic() {        return new ActiveMQTopic("boot-activemq-topic-01");    }}

4. 服务实现类

创建 ActiveMQServiceImpl.java 文件：

package com.hf.boot.service.impl;import com.hf.boot.service.ActiveMQService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.jms.core.JmsMessagingTemplate;import org.springframework.scheduling.annotation.Scheduled;import org.springframework.stereotype.Service;import org.springframework.transaction.annotation.Transactional;import javax.jms.Queue;import javax.jms.Topic;import java.util.UUID;@Service@Transactionalpublic class ActiveMQServiceImpl implements ActiveMQService {    @Autowired    private JmsMessagingTemplate jmsMessagingTemplate;    @Autowired    private Queue queue;    @Autowired    private Topic topic;    @Override    public void sendQueueMsg() {        String message = UUID.randomUUID().toString().replace("-", "").substring(0, 6);        jmsMessagingTemplate.convertAndSend(queue, message);        System.out.println("send queue msg over . . .");    }    @Override    public void sendTopicMsg() {        String message = UUID.randomUUID().toString().replace("-", "").substring(0, 6);        jmsMessagingTemplate.convertAndSend(topic, message);        System.out.println("send topic msg over . . .");    }    @Scheduled(fixedDelay = 3000)    @Override    public void timeSendQueueMsg() {        String message = UUID.randomUUID().toString().replace("-", "").substring(0, 6);        System.out.println("Scheduled推送Queue消息:" + message);        jmsMessagingTemplate.convertAndSend(queue, message);    }    @Scheduled(fixedDelay = 8000)    @Override    public void timeSendTopicMsg() {        String message = UUID.randomUUID().toString().replace("-", "").substring(0, 6);        System.out.println("Scheduled推送Topic消息:" + message);        jmsMessagingTemplate.convertAndSend(topic, message);    }}

5. 消息监听类

创建 ActiveMQListener.java 文件：

package com.hf.boot.listener;import org.springframework.jms.annotation.JmsListener;import org.springframework.stereotype.Component;import javax.jms.JMSException;import javax.jms.TextMessage;@Componentpublic class ActiveMQListener {    @JmsListener(destination = "boot-activemq-queue-01")    public void receiveQueue(Message message) throws JMSException {        System.out.println("消费者接收到queues消息:" + message.getText());    }    @JmsListener(destination = "boot-activemq-topic-01")    public void receiveTopic(Message message) throws JMSException {        System.out.println("消费者接收到topics消息:" + message.getText());    }}

6. 应用程序启动类

创建 BootActivemqApplication.java 文件：

package com.hf.boot;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.scheduling.annotation.EnableScheduling;@SpringBootApplication@EnableSchedulingpublic class BootActivemqApplication {    public static void main(String[] args) {        SpringApplication.run(BootActivemqApplication.class, args);    }}

7. 服务测试说明

• topics 消息测试：首先需要将 application.yml 中的 pub-sub-domain 属性设置为 true，这样才能实现 topics 消息的发布和订阅。如果需要测试 queues 消息，可以将该属性设置为 false。

• queues 消息测试：如需测试 queues 消息的推送和接收，请确保 application.yml 中的 pub-sub-domain 属性设置为 false。

8. 定时任务配置

在 application.yml 中添加定时任务配置：

spring:  jms:    pub-sub-domain: true # true 为 topics，false 为 queues

通过以上配置，可以实现对 topics 和 queues 消息的有效管理和使用。

这样，一个基于 Spring Boot 2.x 整合 ActiveMQ 的消息中间件服务项目就完成了。

转载地址：http://yjnzk.baihongyu.com/