投稿
  1. 数海云首页
  2. 云计算

如何用ActiveMQ实现消息推送？ActiveMQ消息队列使用教程

热舞 • • 云计算 • 阅读 0

ActiveMQ 本身是一个消息中间件（Message Broker），它主要提供点对点（P2P, Queue）和发布/订阅（Pub/Sub, Topic）两种消息模式。

ActiveMQ 并不直接“推送”消息给客户端，而是客户端订阅主题或监听队列，当消息到达时，ActiveMQ 会将消息推送（Push）给已连接的客户端。

以下是使用 ActiveMQ 实现消息推送的完整指南，包括核心概念、Java 代码示例和最佳实践。

核心概念

  1. Producer（生产者）：发送消息的一方。
  2. Consumer（消费者）：接收消息的一方。
  3. Queue（队列）：点对点模式，消息只会被一个消费者消费（负载均衡）。
  4. Topic（主题）：发布/订阅模式，消息会被所有订阅该主题的消费者接收（广播）。

环境准备

引入依赖（Maven）

<dependencies>
    <!-- ActiveMQ Client -->
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-client</artifactId>
        <version>5.17.6</version>
    </dependency>
    <!-- SLF4J for logging -->
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.36</version>
    </dependency>
</dependencies>

启动 ActiveMQ

确保 ActiveMQ 服务已启动，默认连接地址为：tcp://localhost:61616

场景一：点对点模式（Queue） 实现任务分发推送

适用于：订单处理、短信发送等需要每个消息只被处理一次的场景。

生产者（Producer）

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.;
public class QueueProducer {
    public static final String BROKER_URL = "tcp://localhost:61616";
    public static final String QUEUE_NAME = "my-queue";
    public static void main(String[] args) throws JMSException {
        // 1. 创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                BROKER_URL
        );
        // 2. 创建连接
        Connection connection = connectionFactory.createConnection();
        connection.start();
        // 3. 创建会话（非事务，自动确认）
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 4. 创建目标队列
        Destination destination = session.createQueue(QUEUE_NAME);
        // 5. 创建消息生产者
        MessageProducer producer = session.createProducer(destination);
        // 6. 发送消息
       
如何用ActiveMQ实现消息推送？ActiveMQ消息队列使用教程
 for (int i = 0; i < 5; i++) {
            TextMessage message = session.createTextMessage("推送消息内容: " + i);
            producer.send(message);
            System.out.println("已发送消息: " + message.getText());
        }
        // 7. 关闭资源
        producer.close();
        session.close();
        connection.close();
    }
}

消费者（Consumer） 监听模式

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.;
public class QueueConsumer {
    public static final String BROKER_URL = "tcp://localhost:61616";
    public static final String QUEUE_NAME = "my-queue";
    public static void main(String[] args) throws JMSException {
        // 1. 创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                BROKER_URL
        );
        // 2. 创建连接
        Connection connection = connectionFactory.createConnection();
        connection.start(); // 启动连接
        // 3. 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 4. 创建目标队列
        Destination destination = session.createQueue(QUEUE_NAME);
        // 5. 创建消息消费者
        MessageConsumer consumer = session.createConsumer(destination);
        // 6. 设置消息监听器（核心：实现推送效果）
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    if (message instanceof TextMessage) {
                        TextMessage textMessage = (TextMessage) message;
                        System.out.println("收到消息: " + textMessage.getText());
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("消费者已启动，等待消息推送...");
        // 保持主线程运行，实际项目中通常由 Spring 容器管理生命周期
        Thread.sleep(1000000);
        // 7. 关闭资源
        consumer.close();
        session.close();
        connection.close();
    }
}

场景二：发布/订阅模式（Topic） 实现广播推送

适用于：新闻推送、实时通知、股票行情等需要所有订阅者都收到消息的场景。

注意：Topic 模式下，消费者必须在消息发送前先订阅，否则收不到消息。

生产者（Producer）

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.;
public class TopicProducer {
    public static final String BROKER_URL = "tcp://localhost:61616";
    public stat
如何用ActiveMQ实现消息推送？ActiveMQ消息队列使用教程
ic final String TOPIC_NAME = "my-topic";
    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                BROKER_URL
        );
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic(TOPIC_NAME);
        MessageProducer producer = session.createProducer(destination);
        // 发送消息
        TextMessage message = session.createTextMessage("广播消息: 系统更新通知");
        producer.send(message);
        System.out.println("已发送广播消息: " + message.getText());
        producer.close();
        session.close();
        connection.close();
    }
}

消费者（Consumer）

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.;
public class TopicConsumer {
    public static final String BROKER_URL = "tcp://localhost:61616";
    public static final String TOPIC_NAME = "my-topic";
    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD,
                BROKER_URL
        );
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Destination destination = session.createTopic(TOPIC_NAME);
        // 创建消费者并订阅主题
        MessageConsumer consumer = session.createConsumer(destination);
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    if (message instanceof TextMessage) {
                        System.out.println("消费者收到广播: " + ((TextMessage) message).getText());
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        System.out.println("Topic 消费者已启动，等待广播推送...");
        Thread.sleep(1000000);
        consumer.close();
        session.close();
        connection.close();
    }
}

进阶：使用 Spring Boot 简化开发

在实际企业中,通常使用 Spring Boot + Spring JMS 来管理 ActiveMQ，代码更简洁且支持自动重连、事务管理等。

添加依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>activemq-pool</artifactId>
</dependency>

如何用ActiveMQ实现消息推送？ActiveMQ消息队列使用教程

配置文件 application.yml

spring:
  activemq:
    broker-url: tcp://localhost:61616
    user: admin
    password: admin
    in-memory: true
    pool:
      enabled: true # 启用连接池

生产者（使用 JmsTemplate）

import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;
import javax.jms.Destination;
import javax.jms.TextMessage;
@Service
public class QueueSender {
    private final JmsTemplate jmsTemplate;
    private final Destination destination;
    public QueueSender(JmsTemplate jmsTemplate, Destination destination) {
        this.jmsTemplate = jmsTemplate;
        this.destination = destination;
    }
    public void send(String message) {
        jmsTemplate.convertAndSend(destination, message);
    }
}

消费者（使用 @JmsListener）

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
@Component
public class QueueReceiver {
    @JmsListener(destination = "my-queue")
    public void receiveMessage(String message) {
        System.out.println("Spring Boot 消费者收到: " + message);
    }
}

关键注意事项

  1. 持久化 vs 非持久化：

    • 默认情况下,ActiveMQ 会持久化消息，如果消费者离线，消息会保存在 Broker 中，消费者上线后仍能收到。
    • 如果需要高性能且允许丢失消息,可以设置 DeliveryMode.NON_PERSISTENT。

  2. 连接管理：

    • 在生产环境中,不要每次发送消息都创建新的 Connection，应使用连接池（如 ActiveMQConnectionFactory + PooledConnectionFactory）。

  3. ACK 模式：

    • AUTO_ACKNOWLEDGE：自动确认，消息被接收后立即从队列删除。
    • CLIENT_ACKNOWLEDGE：客户端手动确认 message.acknowledge()，适合需要处理成功后才删除消息的场景。
    • DUPS_OK_ACKNOWLEDGE：允许重复消息，适合对一致性要求不高的场景，性能更高。

  4. 安全性：

    • 确保 ActiveMQ 的默认账号密码已修改。
    • 在生产环境中,建议使用 SSL/TLS 加密连接。
  • Queue：适合一对一任务分发，每个消息只被一个消费者处理。
  • Topic：适合一对多广播，所有订阅者都能收到消息。
  • Spring Boot：推荐使用 @JmsListener 和 JmsTemplate 简化开发。

通过以上方式,你可以利用 ActiveMQ 实现高效、可靠的消息推送服务。

【版权声明】：本站所有内容均来自网络，若无意侵犯到您的权利，请及时与我们联系将尽快删除相关内容!

赞 (0)
热舞的头像热舞
0 0
上一篇 2026-07-10 16:27
下一篇 2024-08-19 09:10

相关推荐

  • 国外云计算体系结构到底是什么，国外云计算架构有哪些层次 云计算

    国外云计算体系结构到底是什么，国外云计算架构有哪些层次

    国外云计算体系结构本质上是一套分层的、标准化的技术架构模型，它将庞大的IT资源通过虚拟化技术池化，并以服务的形式按需交付给用户，其核心逻辑在于解耦与抽象，即通过底层硬件的抽象化，向上层应用提供弹性的、自动化的资源调度能力，这种体系结构不仅仅是硬件的堆砌，更是一种从物理设施到应用服务的全栈式交付体系，它彻底改变了……

    热舞的头像 热舞
    2026-04-11
    005
  • 本地导入的数据库文件用什么软件打开？ 云计算

    本地导入的数据库文件用什么软件打开？

    在本地导入数据库文件是许多开发人员和数据库管理员日常工作中的重要环节，无论是备份数据的恢复、跨环境的数据迁移，还是测试数据的初始化，都需要掌握正确的操作方法，不同类型的数据库文件（如MySQL的.sql、.dump，PostgreSQL的.backup，SQLite的.db等）和不同的数据库管理系统（DBMS……

    热舞的头像 热舞
    2025-09-30
    0011
  • 如何高效管理裸金属服务器的云硬盘资源？ 云计算

    如何高效管理裸金属服务器的云硬盘资源？

    服务器云硬盘是一种高性能、可扩展的存储解决方案，适用于裸金属服务器。通过有效的管理策略，可以确保数据的安全性和可靠性，同时提高存储资源的利用率和灵活性。

    热舞的头像 热舞
    2024-08-12
    0010
  • 服务器操作系统可以一键还原吗 云计算

    服务器操作系统可以一键还原吗

    服务器操作系统可通过备份工具或镜像恢复实现类似“一键还原”，但需谨慎操作以避免数据丢失，生产环境建议优先使用专业备份

    热舞的头像 热舞
    2025-05-07
    0015

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

广告合作

QQ：14239236

在线咨询： QQ交谈

邮件：asy@cxas.com

工作时间：周一至周五，9:30-18:30，节假日休息

关注微信