ActiveMQ结合Spring收发消息报错怎么办？Spring整合ActiveMQ配置教程

ActiveMQ 结合 Spring 收发消息示例

下面提供一个完整的示例，展示如何使用 Spring 集成 ActiveMQ 进行点对点（Queue）和发布/订阅（Topic）模式的消息收发。

---

Maven 依赖

<dependencies>
    <!-- Spring JMS -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>5.3.20</version>
    </dependency>
    <!-- ActiveMQ Client -->
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.17.0</version>
    </dependency>
    <!-- Spring Context -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-context</artifactId>
        <version>5.3.20</version>
    </dependency>
    <!-- Spring Beans -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-beans</artifactId>
        <version>5.3.20</version>
    </dependency>
</dependencies>

---

Spring 配置文件（applicationContext-jms.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"
       xmlns:jms="http://www.springframework.org/schema/jms"
       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
           http://www.springframework.org/schema/jms
           http://www.springframework.org/schema/jms/spring-jms.xsd">
    <!-- 1. 连接工厂 -->
    <bean id="connectionFactory"
          class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
        <!-- 可选：设置用户名密码 -->
        <!--
        <property name="userName" value="admin"/>
        <property name="password" value="admin"/>
        -->
    </bean>
    <!-- 2. Spring JMS Template（用于发送消息） -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <!-- 非持久化订阅（Topic 模式需要） -->
        <property name="pubSubDomain" value="false"/>
        <!-- 是否使用默认目的地 -->
        <property name="defaultDestination" ref=&quo
t;myQueue"/>
    </bean>
    <!-- 3. 消息监听容器（用于接收消息） -->
    <bean id="jmsContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="myQueue"/>
        <property name="messageListener" ref="myMessageListener"/>
        <!-- 并发消费者数量 -->
        <property name="concurrency" value="1-5"/>
    </bean>
    <!-- 4. 定义队列目的地 -->
    <bean id="myQueue"
          class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg value="my.queue"/>
    </bean>
    <!-- 5. 定义主题目的地（可选，用于发布/订阅） -->
    <bean id="myTopic"
          class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg value="my.topic"/>
    </bean>
    <!-- 6. 自定义消息监听器 -->
    <bean id="myMessageListener" class="com.example.jms.MyMessageListener"/>
</beans>

---

消息监听器（接收端）

package com.example.jms;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
public class MyMessageListener implements MessageListener {
    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMessage = (TextMessage) message;
                String text = textMessage.getText();
                System.out.println("收到消息: " + text);
            } else {
                System.out.println("收到非文本消息: " + message);
            }
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

---

消息发送服务

package com.example.jms;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
import javax.jms.Destination;
@Service
public class MessageSender {
    @Autowired
    private JmsTemplate jmsTemplate;
    /
      发送消息到队列
     /
    public void sendToQueue(String message) {
        jmsTemplate.send("my.queue", session -> session.createTextMessage(message));
        System.out.println("已发送消息到队列: " + message);
    }
    /
      发送消息到主题（发布/订阅模式）
     /
    public void sendToTopic(String message) {
        jmsTemplate.setPubSubDomain(true);
        jmsTemplate.send("my.topic", session -> session.createTextMessage(message));
        jmsTemplate.setPubSubDomain(false); // 恢复默认
        System.out.println("已发送消息到主题: " + message);
    }
    /
      使用 Destination 对象发送
     /
    public void sendWithDestination(Destination destination, String message) {
        jmsTemplate.send(destination, session -> session.createTextMessage(message));
        System.out.println("已发送消息到目的地: " + destination);
    }
}

---

主程序测试

package com.example;
import com.example.jms.MessageSender;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class JmsDemo {
    public static void main(String[] args) throws InterruptedException {
        // 加载 Spring 配置
        ApplicationContext context = new ClassPathXmlApplicationContext("applicationContext-jms.xml");
        // 获取发送服务
        MessageSender sender = context.getBean(MessageSender.class);
        // 发送消息
        sender.sendToQueue("Hello, ActiveMQ via Spring!");
        sender.sendToQueue("Second message");
        // 等待一段时间让监听器处理消息
        Thread.sleep(5000);
        System.out.println("测试完成");
    }
}

---

使用 Java Config 方式（推荐）

JMS 配置类

package com.example.config;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;
import javax.jms.ConnectionFactory;
import javax.jms.Queue;
import javax.jms.Topic;
@Configuration
public class JmsConfig {
    @Bean
    public ConnectionFactory connectionFactory() {
        ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
        factory.setBrokerURL("tcp://localhost:61616");
        return factory;
    }
    @Bean
    public JmsTemplate jmsTemplate(ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate(connectionFactory);
        template.setDefaultDestination(queue());
        return template;
    }
    @Bean
    public Queue queue() {
        return new ActiveMQQueue("my.queue");
    }
    @Bean
    public Topic topic() {
        return new ActiveMQTopic("my.topic");
    }
    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrency("1-5");
        return factory;
    }
}

使用 @JmsListener 注解接收消息

package com.example.jms;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
@Component
public class AnnotatedMessageListener {
    @JmsListener(destination = "my.queue"
;, containerFactory = "jmsListenerContainerFactory")
    public void receiveMessage(String message) {
        System.out.println("注解方式收到消息: " + message);
    }
    @JmsListener(destination = "my.topic", containerFactory = "jmsListenerContainerFactory")
    public void receiveFromTopic(String message) {
        System.out.println("注解方式从主题收到消息: " + message);
    }
}

发送端（使用 JmsTemplate）

package com.example.service;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
@Service
public class AnnotatedMessageSender {
    @Autowired
    private JmsTemplate jmsTemplate;
    public void send(String message) {
        jmsTemplate.convertAndSend("my.queue", message);
    }
    public void sendToTopic(String message) {
        jmsTemplate.setPubSubDomain(true);
        jmsTemplate.convertAndSend("my.topic", message);
        jmsTemplate.setPubSubDomain(false);
    }
}

---

架构关系图

┌─────────────┐       JmsTemplate       ┌──────────┐
│  Sender      │ ──────────────────────► │          │
│  (Service)   │                         │ ActiveMQ │
└─────────────┘                         │ Broker   │
                                        │ (Queue/  │
┌─────────────┐       @JmsListener      │  Topic)  │
│  Listener   │ ◄────────────────────── │          │
│  (Component)│                         └──────────┘
└─────────────┘

---

关键要点

要点	说明
Queue 模式	点对点，消息只被一个消费者消费
Topic 模式	发布/订阅，所有订阅者都能收到消息
JmsTemplate	用于发送消息，简化 JMS API 调用
@JmsListener	注解方式声明消息监听器，更简洁
DefaultJmsListenerContainerFactory	配置监听容器，支持并发消费
持久化 vs 非持久化	通过 DeliveryMode 控制，默认非持久化
事务管理	可通过 TransactionManager 配置 JMS 事务

提示：确保 ActiveMQ 服务已启动（默认端口 61616），并创建好对应的 Queue 或 Topic，Spring Boot 项目中可使用 spring-boot-starter-activemq starters 进一步简化配置。