springboot整合阿里云的消息队列MQ

简介: springboot整合阿里云的消息队列MQ

1、概述


公司使用的阿里云消息队列MQ服务,框架是springboot。做了一个demo,记录整合过程。


2、步骤


第一步:配置

配置工作基本上就是按照阿里云消息队列MQ的文档所述那样,在控制台的消息队列MQ里面进行配置.快速入门概述

这里有一点注意的点:我们的topic在授权的时候,可以授权给子账号的。(阿里的rocketmq 支持子账号)


第二步:编码

这里涉及到两个知识点:


配置类:

public class BusMqConfig {
    @Value("${mq.topic.business}")
    private String topic;
    @Value("${mq.producerId.business}")
    private String producerId;
    @Value("${mq.consumerId.business}")
    private String consumerId;
    @Value("${mq.accesskey}")
    private String accesskey;
    @Value("${mq.secretkey}")
    private String secretkey;
    @Value("${mq.onsaddr}")
    private String onsaddr;
    @Value("${mq.subExpression}")
    private String subExpression;
    public String getSubExpression() {
        return subExpression;
    }
    public void setSubExpression(String subExpression) {
        this.subExpression = subExpression;
    }
    //提供消费者的配置
    public Properties getConsumerProperties() {
        Properties consumerProperties = new Properties();
        consumerProperties.setProperty(PropertyKeyConst.ConsumerId, consumerId);
        consumerProperties.setProperty(PropertyKeyConst.AccessKey, accesskey);
        consumerProperties.setProperty(PropertyKeyConst.SecretKey, secretkey);
        consumerProperties.setProperty(PropertyKeyConst.ONSAddr, onsaddr);
        return consumerProperties;
    }
    //提供生产者的配置
    public Properties getProducerProperties() {
        Properties producerProperties = new Properties();
        producerProperties.setProperty(PropertyKeyConst.ProducerId, producerId);
        producerProperties.setProperty(PropertyKeyConst.AccessKey, accesskey);
        producerProperties.setProperty(PropertyKeyConst.SecretKey, secretkey);
        producerProperties.setProperty(PropertyKeyConst.ONSAddr, onsaddr);
        return producerProperties;
    }
    public String getTopic() {
        return topic;
    }
    public void setTopic(String topic) {
        this.topic = topic;
    }
}
复制代码


消费者:在容器启动后立马开始消费

public class MqConsumer implements InitializingBean, DisposableBean{
    @Autowired
    BusMqConfig busMqConfig;
    private Consumer busConsumer;
    @Override
    public void afterPropertiesSet() throws Exception {
        System.out.println("消费者初始化");
        busConsumer = ONSFactory.createConsumer(busMqConfig.getConsumerProperties());
       // busConsumer.start();
        System.out.println("消费者初始化完成");
    }
    public void start(){
        busConsumer.start();
    }
    public void onMessage(){
        busConsumer.subscribe(busMqConfig.getTopic(), busMqConfig.getSubExpression(), new MessageListener() {
            @Override
            public Action consume(Message message, ConsumeContext context) {
               // System.out.println(JSON.toJSONString(message));
                System.out.println("Receive: " + message);
                System.out.println(new String(message.getBody()));
                return Action.CommitMessage;
            }
        });
    }
    @Override
    public void destroy() throws Exception {
        busConsumer.shutdown();
        System.out.println("停止");
    }
}
复制代码


配置容器启动开始消费:

@Autowired
    MqConsumer mqConsumer;
    @Override
    public void run(String... strings) throws Exception {
        System.out.println("开始消费");
        mqConsumer.start();
        mqConsumer.onMessage();
    }

此时我们先在控制台上,用阿里提供的界面化的窗口,模拟发送几条消息。此时的消息的状态是未消费状态。启动项目后,就会在控制台看到消费记录了。


生产者

private final static Logger LOGGER = LoggerFactory.getLogger(MqProducer.class);
    @Autowired
    BusMqConfig busMqConfig;
    private Producer producer;
    @Override
    public void afterPropertiesSet() throws Exception {
        System.out.println("生产者初始化");
        producer = ONSFactory.createProducer(busMqConfig.getProducerProperties());
        producer.start();
    }
    public void sentMessage(Message message){
            producer.sendAsync(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    LOGGER.info(sendResult.getTopic()+"-----"+sendResult.getMessageId());
                }
                @Override
                public void onException(OnExceptionContext context) {
                    LOGGER.error(context.getTopic()+"-----"+context.getMessageId()+":error="+context.getException());
                }
            });
    }
    @Override
    public void destroy() throws Exception {
            producer.shutdown();
    }


创建一个controller,通过postman向接口发送消息体。

@PostMapping(value = "/send")
    public Message msg(@RequestBody String msgbody, HttpServletRequest request){
        System.out.println(msgbody);
        Message msg = new Message(
                // Message 所属的 Topic
                busMqConfig.getTopic(),
                // Message Tag,
                // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在 MQ 服务器过滤
                "TagA",
                // Message Body
                // 任何二进制形式的数据,MQ 不做任何干预,需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式
                msgbody.getBytes());
        // 设置代表消息的业务关键属性,请尽可能全局唯一。
        // 以方便您在无法正常收到消息情况下,可通过阿里云服务器管理控制台查询消息并补发。
        // 注意:不设置也不会影响消息正常收发
        msg.setKey("ORDERID_" + Math.round(Math.random()*8999+1000));
        mqProducer.sentMessage(msg);
        return msg;
    }

至此demo就结束。其实过程很简单


3、总结


此次整合的过程。看似简单。我感觉还有深层的东西在里面

  • spring中的bean生命周期的各个阶段,我们是否能记住,并运用了。我们整天吵着读源码,是否真的读了?读了是否有会运用?我想这就是研究spring架构的意义吧。
  • springboot 虽然使得开发人员的工作量减少了。但是那些关键点,我们是否了解,并掌握。
  • 当我们项目中要整合新的东西时,头绪在哪? 答:官方文档+源码+别人的理解=自己的理解


4、扩展


此demo某些部分还是有其他替代方案的

  • 应用启动后立即开始消费消息:可以创建一个监听实现ApplicationListener
  • 消费者生产者:我是自定义类,把阿里提供的custemer producer 做了封装。其实我们还可以直接将阿里的custemer producer作为bean配置。但我觉得封装下还是比较灵活的

demo地址

网格学习法:由一个点,或者一个项目将各个知识点串起来,锚点连接,由点到面,串成网格。


相关实践学习
快速体验阿里云云消息队列RocketMQ版
本实验将带您快速体验使用云消息队列RocketMQ版Serverless系列实例进行获取接入点、创建Topic、创建订阅组、收发消息、查看消息轨迹和仪表盘。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
3月前
|
消息中间件 数据管理 Serverless
阿里云消息队列 Apache RocketMQ 创新论文入选顶会 ACM FSE 2025
阿里云消息团队基于 Apache RocketMQ 构建 Serverless 消息系统,适配多种主流消息协议(如 RabbitMQ、MQTT 和 Kafka),成功解决了传统中间件在可伸缩性、成本及元数据管理等方面的难题,并据此实现 ApsaraMQ 全系列产品 Serverless 化,助力企业提效降本。
|
17天前
|
消息中间件 Ubuntu Java
SpringBoot整合MQTT实战:基于EMQX实现双向设备通信
本教程指导在Ubuntu上部署EMQX 5.9.0并集成Spring Boot实现MQTT双向通信,涵盖服务器搭建、客户端配置及生产实践,助您快速构建企业级物联网消息系统。
174 1
|
25天前
|
消息中间件 Java Kafka
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例,并结合 Spring 生态的集成方式,帮助开发者根据实际需求选择合适的消息中间件,提升系统解耦、可扩展性与可靠性。
消息队列比较:Spring 微服务中的 Kafka 与 RabbitMQ
|
4月前
|
监控 安全 Java
Java 开发中基于 Spring Boot 3.2 框架集成 MQTT 5.0 协议实现消息推送与订阅功能的技术方案解析
本文介绍基于Spring Boot 3.2集成MQTT 5.0的消息推送与订阅技术方案,涵盖核心技术栈选型(Spring Boot、Eclipse Paho、HiveMQ)、项目搭建与配置、消息发布与订阅服务实现,以及在智能家居控制系统中的应用实例。同时,详细探讨了安全增强(TLS/SSL)、性能优化(异步处理与背压控制)、测试监控及生产环境部署方案,为构建高可用、高性能的消息通信系统提供全面指导。附资源下载链接:[https://pan.quark.cn/s/14fcf913bae6](https://pan.quark.cn/s/14fcf913bae6)。
581 0
|
4月前
|
消息中间件 存储 大数据
阿里云消息队列 Kafka 架构及典型应用场景
阿里云消息队列 Kafka 是一款基于 Apache Kafka 的分布式消息中间件,支持消息发布与订阅模型,满足微服务解耦、大数据处理及实时流数据分析需求。其通过存算分离架构优化成本与性能,提供基础版、标准版和专业版三种 Serverless 版本,分别适用于不同业务场景,最高 SLA 达 99.99%。阿里云 Kafka 还具备弹性扩容、多可用区部署、冷热数据缓存隔离等特性,并支持与 Flink、MaxCompute 等生态工具无缝集成,广泛应用于用户行为分析、数据入库等场景,显著提升数据处理效率与实时性。
|
12月前
|
消息中间件 JSON Java
开发者如何使用轻量消息队列MNS
【10月更文挑战第19天】开发者如何使用轻量消息队列MNS
771 102
|
12月前
|
消息中间件 安全 Java
云消息队列RabbitMQ实践解决方案评测
一文带你详细了解云消息队列RabbitMQ实践的解决方案优与劣
284 112
|
10月前
|
消息中间件 监控 Java
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
如何将Spring Boot + RabbitMQ应用程序部署到Pivotal Cloud Foundry (PCF)
148 6

相关产品

  • 云消息队列 MQ