AI 助理
文档备案控制台
开发者社区 开发与运维 文章 正文

10 Spring集成MQTT（生产者与消费者）

2023-11-03 356
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 10 Spring集成MQTT（生产者与消费者）

代码已上传至我的Github仓库：https://github.com/ylw-github/Spring-MQTT-Demo.git

整个代码的结构：

生产者（producer）

依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>com.ylw</groupId>
        <artifactId>spring-mqtt-demo</artifactId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <artifactId>producer</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-core</artifactId>
            <version>4.3.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <version>4.3.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.0</version>
        </dependency>
    </dependencies>
</project>

配置文件（mqtt_producer.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:int="http://www.springframework.org/schema/integration"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
       xsi:schemaLocation="
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-4.1.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
        http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt-4.1.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-4.1.xsd">
    <bean id="clientFactory"
          class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
        <property name="userName" value="user1"/>
        <property name="password" value="123456"/>
        <property name="serverURIs">
            <array>
                <value>tcp://127.0.0.1:1883</value>
            </array>
        </property>
    </bean>
    <bean id="mqtt" class="org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler">
        <constructor-arg name="clientId" value="clientId_producer"></constructor-arg>
        <constructor-arg name="clientFactory" ref="clientFactory"></constructor-arg>
    </bean>
</beans>

单元测试类（ProducerTestUnit.java）：

package com.ylw;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:mqtt_producer.xml")
public class ProducerTestUnit {
    @Autowired
    private MqttPahoMessageHandler mqtt;
    @Test
    public void sendTextQueue() {
        sendMqttMsg("testTopic", 2, "hello world send...");
    }
    /**
     * @param topicName 主题名字
     * @param message   发送的消息
     * @author YangLinWei
     * @date 2019/9/4  10:17
     * @qos 请求服务质量，0：至多一次，1：至少一次，2：刚好一次
     */
    private void sendMqttMsg(String topicName, int qos, String message) {
        Message<String> messages = MessageBuilder.withPayload(message).setHeader(MqttHeaders.TOPIC, topicName)
                .setHeader(MqttHeaders.QOS, qos)
                .setHeader(MqttHeaders.RETAINED, true).build();
        mqtt.handleMessage(messages);
    }
}

消费者（consumer）：

依赖：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>spring-mqtt-demo</artifactId>
        <groupId>com.ylw</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>consumer</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-core</artifactId>
            <version>4.3.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-mqtt</artifactId>
            <version>4.3.9.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.2.0</version>
        </dependency>
    </dependencies>
</project>

配置文件（mqtt_consumer.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:int="http://www.springframework.org/schema/integration"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:int-mqtt="http://www.springframework.org/schema/integration/mqtt"
       xsi:schemaLocation="
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-4.1.xsd
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.1.xsd
        http://www.springframework.org/schema/integration/mqtt http://www.springframework.org/schema/integration/mqtt/spring-integration-mqtt-4.1.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context-4.1.xsd">
    <bean id="mqttCallback" class="com.ylw.MqttReciever"></bean>
    <bean id="options" class="org.eclipse.paho.client.mqttv3.MqttConnectOptions">
        <property name="cleanSession" value="true"></property>
        <property name="connectionTimeout" value="30"></property>
        <property name="keepAliveInterval" value="45"></property>
        <property name="userName" value="user1"></property>
        <property name="password" value="123456"></property>
    </bean>
    <bean id="client" class="org.eclipse.paho.client.mqttv3.MqttClient">
        <constructor-arg name="serverURI" value="tcp://127.0.0.1:1883"></constructor-arg>
        <constructor-arg name="clientId" value="clientId_consumer"></constructor-arg>
        <property name="callback" ref="mqttCallback"></property>
    </bean>
</beans>

单元测试类（ConsumerTestUnit.java）：

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:mqtt_consumer.xml")
public class ConsumerTestUnit {
    @Autowired
    MqttClient client;
    @Autowired
    private MqttConnectOptions options;
    @Test
    public void testQueue() {
        try {
            client.connect(options);
            client.subscribe("testTopic", 2);
            System.in.read();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消息监听类（MqttReciever.java）：

package com.ylw;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
public class MqttReciever implements MqttCallback {
    @Override
    public void connectionLost(Throwable throwable) {
    }
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        System.out.println("Client 接收消息主题 : " + topic);
        System.out.println("Client 接收消息Qos : " + message.getQos());
        System.out.println("Client 接收消息内容 : " + new String(message.getPayload()));
    }
    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
    }
}
文章标签：
云消息队列 MQ
Java
Spring
测试技术
关键词：
Spring集成
云消息队列 MQ spring
云消息队列 MQ集成
集成spring
云消息队列 MQ消费者
相关实践学习
快速体验阿里云云消息队列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
阿甘兄
目录
相关文章
HELLO程序员
|
9月前
|
数据可视化 Java BI
将 Spring 微服务与 BI 工具集成：最佳实践
本文探讨了 Spring 微服务与商业智能（BI）工具集成的潜力与实践。随着微服务架构和数据分析需求的增长，Spring Boot 和 Spring Cloud 提供了构建可扩展、弹性服务的框架，而 BI 工具则增强了数据可视化与实时分析能力。文章介绍了 Spring 微服务的核心概念、BI 工具在企业中的作用，并深入分析了两者集成带来的优势，如实时数据处理、个性化报告、数据聚合与安全保障。同时，文中还总结了集成过程中的最佳实践，包括事件驱动架构、集中配置管理、数据安全控制、模块化设计与持续优化策略，旨在帮助企业构建高效、智能的数据驱动系统。
HELLO程序员
426 1
将 Spring 微服务与 BI 工具集成：最佳实践
互联网课堂
|
11月前
|
XML 人工智能 Java
Spring Boot集成Aviator实现参数校验
Aviator是一个高性能、轻量级的Java表达式求值引擎，适用于动态表达式计算。其特点包括支持多种运算符、函数调用、正则匹配、自动类型转换及嵌套变量访问，性能优异且依赖小。适用于规则引擎、公式计算和动态脚本控制等场景。本文介绍了如何结合Aviator与AOP实现参数校验，并附有代码示例和仓库链接。
互联网课堂
667 0
糊涂糊涂123
|
11月前
|
安全 Java 数据库
第16课：Spring Boot中集成 Shiro
第16课：Spring Boot中集成 Shiro
糊涂糊涂123
1128 0
糊涂糊涂123
|
11月前
|
消息中间件 存储 Java
第15课： Spring Boot中集成ActiveMQ
第15课： Spring Boot中集成ActiveMQ
糊涂糊涂123
668 0
HELLO程序员
|
9月前
|
消息中间件 Java Kafka
消息队列比较：Spring 微服务中的 Kafka 与 RabbitMQ
本文深入解析了 Kafka 和 RabbitMQ 两大主流消息队列在 Spring 微服务中的应用与对比。内容涵盖消息队列的基本原理、Kafka 与 RabbitMQ 的核心概念、各自优势及典型用例，并结合 Spring 生态的集成方式，帮助开发者根据实际需求选择合适的消息中间件，提升系统解耦、可扩展性与可靠性。
HELLO程序员
612 1
消息队列比较：Spring 微服务中的 Kafka 与 RabbitMQ
HELLO程序员
|
9月前
|
消息中间件 存储 Java
RabbitMQ 和 Spring Cloud Stream 实现异步通信
本文介绍了在微服务架构中，如何利用 RabbitMQ 作为消息代理，并结合 Spring Cloud Stream 实现高效的异步通信。内容涵盖异步通信的优势、RabbitMQ 的核心概念与特性、Spring Cloud Stream 的功能及其与 RabbitMQ 的集成方式。通过这种组合，开发者可以构建出具备高可用性、可扩展性和弹性的分布式系统，满足现代应用对快速响应和可靠消息传递的需求。
HELLO程序员
470 2
RabbitMQ 和 Spring Cloud Stream 实现异步通信
JJLIN距离
|
9月前
|
监控 Cloud Native Java
Spring Integration 企业集成模式技术详解与实践指南
本文档全面介绍 Spring Integration 框架的核心概念、架构设计和实际应用。作为 Spring 生态系统中的企业集成解决方案，Spring Integration 基于著名的 Enterprise Integration Patterns（EIP）提供了轻量级的消息驱动架构。本文将深入探讨其消息通道、端点、过滤器、转换器等核心组件，以及如何构建可靠的企业集成解决方案。
JJLIN距离
787 0
盹猫
|
11月前
|
Java 关系型数据库 数据库连接
Spring Boot项目集成MyBatis Plus操作PostgreSQL全解析
集成 Spring Boot、PostgreSQL 和 MyBatis Plus 的步骤与 MyBatis 类似，只不过在 MyBatis Plus 中提供了更多的便利功能，如自动生成 SQL、分页查询、Wrapper 查询等。
盹猫
1018 2
糊涂糊涂123
|
11月前
|
缓存 JSON 前端开发
第07课：Spring Boot集成Thymeleaf模板引擎
第07课：Spring Boot集成Thymeleaf模板引擎
糊涂糊涂123
906 0
第07课：Spring Boot集成Thymeleaf模板引擎

热门文章

最新文章

  • 1
    消息中间件选型分析——从Kafka与RabbitMQ的对比来看全局
  • 2
    RocketMQ重平衡策略你用过几种？
  • 3
    自顶向下学习 RocketMQ（五）：顺序消息原理
  • 4
    springboot整合阿里云的消息队列MQ
  • 5
    【实测】django的超轻量级消息队列：django-task-mq 使用教程
  • 6
    MQTT常见问题之用rocketmq mqttdemo的MqttConsumer始终无法接收到消息如何解决
  • 7
    RocketMQ运维自我实践
  • 8
    RocketMQ平台的消息灰度方案（2）
  • 9
    带你从头进行RabbitMQ安装、集群搭建、镜像队列配置和代码验证（上）
  • 10
    可运维：云原生时代 RocketMQ 运维管控的利器 - RocketMQ Operator：
  • 1
    构建未来：AI驱动的自适应网络安全防御系统提升软件测试效率：自动化与持续集成的实践之路
    1177
  • 2
    深入理解与实施软件测试中的持续集成策略
    371
  • 3
    探索自动化测试在持续集成中的应用与优化策略
    250
  • 4
    SpringBoot集成和使用消息队列
    633
  • 5
    构建高效持续集成系统的策略与实践
    560
  • 6
    深入理解与应用软件测试中的持续集成策略
    388
  • 7
    构建高效自动化运维体系：基于容器技术的持续集成与持续部署实践
    544
  • 8
    深入探究持续集成在软件测试中的应用与优化
    267
  • 9
    构建高效自动化运维系统：基于容器技术的持续集成与持续部署实践
    290
  • 10
    【阿里云弹性计算】阿里云 ECS 与 Kubernetes 集成：轻松管理容器化应用
    758

    • 相关课程

    更多
  • 全面讲解Spring Cloud Alibaba技术栈（知识精讲+项目实战）第三阶段
  • 精通Spring Cloud Alibaba
  • 微服务框架 Spring Cloud 快速入门
  • Java Web开发系列课程 - Spring框架入门
  • Spring Boot 2.5.x开发实战
  • Spring Cloud微服务架构设计与开发实战

    • 相关电子书

    更多
  • Spring Cloud Hoxton 新版本介绍 & 未来展望
  • 集成智能接入网关APP：优化企业级移动办公网络
  • 云效助力企业集成安全到DevOps中
    • 下一篇
    阿里云正式发布 Agentic 代码安全：AI驱动的双Agent协同引擎