Spring Integration + MQTT

1. 简介

Spring Integration：

Spring Integration是一个开源的Java库，用于构建基于消息的应用程序。它提供了一套丰富的组件和工具，使得开发者可以轻松地开发出可靠、灵活和可扩展的集成解决方案。以下是Spring Integration的一些主要用途：

  1. 企业服务总线（ESB）： Spring Integration可以用来构建企业服务总线，它支持各种协议和消息格式，使得不同系统间的数据和事件可以轻松交换。

  2. 消息传递和解耦： 它支持在不同的应用程序组件之间进行异步消息传递，从而降低系统组件间的耦合度。

  3. 事件驱动架构： Spring Integration支持事件驱动的架构风格，允许系统对事件做出响应，而不是基于传统的请求-响应模型。

  4. 数据转换和路由： 提供数据转换和路由的功能，可以将数据从一种格式转换为另一种格式，并根据内容将消息路由到不同的目的地。

  5. 错误处理： 它提供了一套完整的错误处理机制，包括重试、补偿和消息存储等策略。

  6. 文件和数据库集成： 可以轻松地与文件系统和数据库进行集成，支持文件传输、数据库操作等场景。

  7. 外部系统适配： 通过提供各种适配器，Spring Integration可以与外部系统（如JMS、AMQP、HTTP、FTP等）进行集成。

  8. 批处理和任务调度： 支持批处理操作和任务调度，可以用于处理大量数据或定时任务。

  9. 模块化和可扩展性： 它的模块化设计使得开发者可以根据需要添加或替换组件，从而构建高度可扩展的系统。

  10. 多环境支持： 支持多种部署环境，包括本地应用、云环境和微服务架构。

  11. 开发和配置的简化： 通过提供声明式的配置和简化的编程模型，Spring Integration降低了开发复杂性，并缩短了开发周期。

  12. 社区和生态系统： 作为Spring家族的一部分，Spring Integration受益于活跃的社区和广泛的生态系统，提供了大量的资源和支持。

Spring Integration + MQTT：

Spring Integration与MQTT的集成是一个非常强大的组合，它允许开发者在Spring应用程序中轻松地实现MQTT协议的消息发布和订阅功能。以下是Spring Integration与MQTT集成的一些主要用途和优势：

  1. 轻量级消息传递： MQTT是一种轻量级的消息传递协议，特别适合带宽有限和延迟敏感的环境，如物联网（IoT）应用。Spring Integration通过提供MQTT通道适配器，使得在Spring应用程序中集成MQTT变得简单直接 。

  2. 简化配置： 通过Spring Integration，开发者可以使用声明式配置来定义MQTT的入站（订阅）和出站（发布）消息通道，而不需要深入了解MQTT客户端库的复杂性 。

  3. 支持MQTT v5： 从Spring Integration 5.5.5版本开始，支持MQTT v5协议，包括对MQTT v5特有的消息属性的支持，如消息过期间隔、响应主题等 。

  4. 灵活的消息处理： Spring Integration提供了强大的消息处理能力，包括消息转换、路由、聚合、分割等，这些都可以通过声明式配置轻松实现 。

  5. 错误处理和重连机制： Spring Integration提供了错误处理机制，包括请求处理建议，例如重试或断路器。同时，支持MQTT的自动重连机制，确保了消息传递的可靠性 。

  6. 与Spring生态系统的集成： 作为Spring家族的一部分，Spring Integration可以很容易地与其他Spring项目（如Spring Boot、Spring Cloud等）集成，提供了与Spring Security、Spring Data等的无缝集成 。

  7. 提高开发效率： Spring Integration的声明式配置和编程模型简化了消息系统开发，降低了开发复杂性，并缩短了开发周期 。

  8. 动态主题管理： Spring Integration允许在运行时动态添加和删除MQTT订阅主题，提供了更高的灵活性 。

  9. 事件驱动架构： 支持事件驱动的架构风格，允许系统对事件做出响应，而不是基于传统的请求-响应模型 。

2. 基本时序架构

        1. 监听到订阅topic有消息流程

        2. 生产者推送一条消息后，中间经过一系列流程后被消费者消费的完整流程

3. 接收消息

通常涉及以下几个步骤：

1. 配置MQTT连接： 首先，需要配置与MQTT代理（如EMQX）的连接。这通常涉及到配置一个MqttPahoClientFactory Bean，它负责创建和管理MQTT客户端连接。

@Bean
public MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();factory.setConnectionOptions(mqttConnectOptions());return factory;
}

2. 创建入站通道适配器： 使用MqttPahoMessageDrivenChannelAdapter创建一个入站通道适配器。这个适配器负责从MQTT代理订阅主题，并在接收到消息时将消息发送到Spring Integration的通道。

@Slf4j
@Configuration
@IntegrationComponentScan
public class MqttInboundConfiguration {@Autowiredprivate MqttPahoClientFactory mqttClientFactory;@Resource(name = ChannelName.INBOUND)private MessageChannel inboundChannel;/*** Clients of inbound message channels.* @return*/@Bean(name = "adapter")public MessageProducerSupport mqttInbound() {MqttClientOptions options = MqttConfiguration.getBasicClientOptions();MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(options.getClientId() + "_consumer_" + System.currentTimeMillis(),mqttClientFactory, options.getInboundTopic().split(","));DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter();// use byte types uniformlyconverter.setPayloadAsBytes(true);adapter.setConverter(converter);adapter.setQos(1);adapter.setOutputChannel(inboundChannel);// 添加钩子函数，确保在程序关闭时正确断开连接Runtime.getRuntime().addShutdownHook(new Thread(() -> {try {if (adapter != null) {adapter.stop();log.warn("[consumer] MQTT client stopped successfully.");}} catch (Exception e) {log.error("[consumer] MQTT client stopped with error: {}",e.getMessage(),e);}}));return adapter;}

3. 配置消息通道： 配置一个消息通道（如DirectChannel），用于传输从MQTT代理接收到的消息。

@Bean(name = ChannelName.INBOUND)
public MessageChannel inboundChannel() {return new ExecutorChannel(threadPool);
}

4. 设置消息监听器： 使用@ServiceActivator注解定义一个服务激活器，它将作为消息监听器处理接收到的消息。这个消息监听器可以是一个方法，这个方法将对通道中的消息进行处理。

5. 处理消息： 实现业务逻辑来处理消息。这通常涉及到从消息中提取数据，并执行所需的操作，例如更新数据库、调用服务或触发事件。

@Bean
@ServiceActivator(inputChannel = ChannelName.INBOUND)
public MessageHandler defaultInboundHandler() {return message -> {// 处理消息// log.info("The default channel does not handle messages." +//         "\nTopic: " + message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC) +//         "\nPayload: " + message.getPayload());};
}

4. 发布信息

        发送MQTT消息通常是通过配置出站通道适配器（MqttOutboundChannelAdapter）来实现的。这个适配器负责将从Spring Integration通道中传来的消息发布到指定的MQTT主题上。

发送MQTT消息的步骤：

1. 配置MQTT客户端工厂（MqttPahoClientFactory）： 这个工厂负责创建和管理MQTT客户端连接。

@Bean
public MqttPahoClientFactory mqttClientFactory() {DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();factory.setConnectionOptions(mqttConnectOptions());return factory;
}

2. 配置MQTT出站通道适配器（MqttOutboundChannelAdapter）： 这个适配器将消息通道中的消息发布到MQTT代理上。

@Configuration
public class MqttOutboundConfiguration {@Autowiredprivate MqttPahoClientFactory mqttClientFactory;@Bean@ServiceActivator(inputChannel = ChannelName.OUTBOUND)public MqttOutboundChannelAdapter mqttOutboundAdapter() {MqttOutboundChannelAdapter adapter = new MqttOutboundChannelAdapter("client_id", mqttClientFactory, "outputTopic");adapter.setQos(1); // 设置服务质量adapter.setAsync(true); // 异步发送消息return adapter;}
}

可以通过setDefaultTopic方法设置默认主题，这样在发送消息时如果没有指定主题，就会使用这个默认主题。

3. 发送消息到消息通道： 通过编程方式或通过其他Spring Integration组件，将消息发送到与MqttOutboundChannelAdapter绑定的消息通道。

@Autowired
private MessageChannel mqttOutboundChannel;public void sendMqttMessage(String payload) {mqttOutboundChannel.send(MessageBuilder.withPayload(payload).build());
}

注：

1. 要确定消息发送到哪一个主题，可以在发送消息时通过消息头MqttHeaders.TOPIC指定。如果没有指定，就会使用在MqttPahoMessageHandler中配置的默认主题。

@Autowired
private MessageChannel mqttOutboundChannel;public void sendMqttMessage(String topic, String payload) {mqttOutboundChannel.send(MessageBuilder.withPayload(payload).setHeader(MqttHeaders.TOPIC, topic).build());
}

2. 通过使用IMqttMessageGateway接口去发送消息到OUTBOUND通道，再由MqttPahoMessageHandler去处理消息

@Component
@MessagingGateway(defaultRequestChannel = ChannelName.OUTBOUND)
public interface IMqttMessageGateway {/*** Publish a message to a specific topic.* @param topic target* @param payload   message*/void publish(@Header(MqttHeaders.TOPIC) String topic, byte[] payload);/*** Use a specific qos to push messages to a specific topic.* @param topic     target* @param payload   message* @param qos   qos*/void publish(@Header(MqttHeaders.TOPIC) String topic, byte[] payload, @Header(MqttHeaders.QOS) int qos);
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/56276.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

电话聊天狂人

电话聊天狂人

给定大量手机用户通话记录，找出其中通话次数最多的聊天狂人。 输入格式: 输入首先给出正整数N（≤105），为通话记录条数。随后N行，每行给出一条通话记录。简单起见，这里只列出拨出方和接收方的11位数字构成…
阅读更多...
jenkins 插件Publish Over SSH

jenkins 插件Publish Over SSH

一、安装插件 二、配置sshserver http://192.168.31.156:8080/manage/configure 三、添加自由风格：PublishOverSSHDemo 我们将工作目录：/var/lib/jenkins/workspace/PublishOverSSHDemo下的图片m3.jpeg 同步到目标143服务器目录：/root/imag…
阅读更多...
Python脚本爬取目标网站上的所有链接

Python脚本爬取目标网站上的所有链接

一、爬取后txt文件保存 需要先pip install requests和BeautifulSoup库 import requests from bs4 import BeautifulSoup# 定义要爬取的新闻网站URL url https://www.chinadaily.com.cn/ # China Daily 网站# 发送请求获取页面内容 response requests.get(url)# 检查请求是否…
阅读更多...
Jgit的使用

Jgit的使用

Jgit的使用 文章目录 Jgit的使用一，git操作的对应代码1.1 查看操作1.1.1 打开仓库1.1.3 获取状态信息 1.2 添加操作1.2.1 初始化本地仓库1.2.2 创建一个新文件并写入内容1.2.3 添加指定（所有）文件到暂存区1.2.4 提交操作1.2.5 连接并推送到远…
阅读更多...
Android中的三种数据存储方式

Android中的三种数据存储方式

目录 1.文件存储 1）内部存储 1--MODE_PRIVATE: 2--MODE_APPEND: 3--MODE_WORLD_READABLE: 4--MODE_WORLD_WRITEABLE: 5--简单使用 3）外部存储 4）内部读取 4）外部读取 2.SharePreferences存储 1）基本概念 2&#xff09…
阅读更多...
设计模式 - 行为模式

设计模式 - 行为模式

行为模式 观察者模式，策略模式，命令模式，中介者模式，备忘录模式，模板方法模式，迭代器模式，状态模式，责任链模式，解释器模式，访问者模式 保存/封装 行为/请求…
阅读更多...
MySQL表连接查询算法

MySQL表连接查询算法

前言 MySQL属于关系型数据库，我们建的表大多也都存在业务上的关联关系，同时我们又不可能将所有的数据都冗余一份，这不符合数据库的设计范式。因此，当我们需要把多张表的数据融合在一起的时候，就需要使用到「多表连接查…
阅读更多...
CSS伪元素详解

CSS伪元素详解

CSS伪元素详解 一、引言 在CSS中，伪元素是一个非常强大的工具，它允许我们为元素的特定部分添加样式，而无需修改HTML结构。这不仅提高了样式的灵活性，还有助于保持代码的整洁和可维护性。本文将深入探讨CSS伪元素的使用方法和一些…
阅读更多...
详解tar.gz, tar.xz, tar, gz后缀文件的区别

详解tar.gz, tar.xz, tar, gz后缀文件的区别

详解tar.gz, tar.xz,tar, gz后缀文件的区别 tar.gz、tar.xz、tar 和 gz 是常见的文件压缩与归档格式，它们的区别主要在于文件的归档和压缩方式。 1. tar 文件 全称：Tape Archive扩展名：.tar说明：tar 文件本身并没有压缩&#x…
阅读更多...
SQL分类中的DDL

SQL分类中的DDL

DDL（Data Definition Language):数据定义语言，用来定义数据库对象（数据库，表，字段）。 一、DDL语句操作数据库 1、查询所有数据库：show databases；（一般用大写&#xff…
阅读更多...
C语言动态内存开辟

C语言动态内存开辟

文章目录 malloc函数free函数calloc函数realloc函数二维数组的动态内存开辟 malloc函数 malloc函数包含再#include<stdlib.h>头文件中 void* malloc (size_t size);这个函数向内存申请⼀块连续可⽤的空间&#xff0c;并返回指向这块空间的指针。 如果开辟成功&#xff…
阅读更多...
OpenCV-人脸检测

OpenCV-人脸检测

文章目录 一、人脸检测流程二、关键方法三、代码示例四、注意事项 OpenCV是一个开源的计算机视觉和机器学习软件库&#xff0c;它提供了多种人脸检测方法&#xff0c;以下是对OpenCV人脸检测的详细介绍&#xff1a; 一、人脸检测流程 人脸检测是识别图像中人脸位置的过程&…
阅读更多...
【Docker系列】Docker查看镜像架构

【Docker系列】Docker查看镜像架构

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…
阅读更多...
模态与非模态的对话框

模态与非模态的对话框

本文学习自&#xff1a; 《Qt Creato快速入门》 #include "widget.h" #include <QApplication>int main(int argc, char *argv[]) {QApplication a(argc, argv);Widget w;w.show();return a.exec(); }1. #include "widget.h" #include "ui_w…
阅读更多...
MySQL数据的导入

MySQL数据的导入

【图书推荐】《MySQL 9从入门到性能优化&#xff08;视频教学版&#xff09;》-CSDN博客 《MySQL 9从入门到性能优化&#xff08;视频教学版&#xff09;&#xff08;数据库技术丛书&#xff09;》(王英英)【摘要 书评 试读】- 京东图书 (jd.com) MySQL9数据库技术_夏天又到了…
阅读更多...
小白也能学会的预测新模型！ReliefF特征选择+XGBoost回归！

小白也能学会的预测新模型！ReliefF特征选择+XGBoost回归！

小白也能学会的预测新模型&#xff01;ReliefF特征选择XGBoost回归&#xff01; 目录 小白也能学会的预测新模型&#xff01;ReliefF特征选择XGBoost回归&#xff01;预测效果基本介绍程序设计参考资料 预测效果 基本介绍 Matlab实现ReliefF-XGBoost多变量回归预测 1.excel数据…
阅读更多...
linux应用

linux应用

检查Python程序未运行则重新运行 entity_program定时杀掉进程重新运行 match_program定时检查是否运行&#xff0c;未运行则启动 (注意echo时间时&#xff0c;date和中间要有空格) #!/bin/bash# 检测的Python程序名称 entity_program"entity.py" match_program"…
阅读更多...
算法收敛的一些证明方法与案例

算法收敛的一些证明方法与案例

证明一个算法收敛通常涉及多个角度&#xff0c;以下是一些常用的方法和示例&#xff1a; 一、方法 1. 数学归纳法 通过数学归纳法证明算法在每一步的输出结果都在收敛范围内。 示例&#xff1a;考虑一个递归算法&#xff0c;假设我们要证明它在每一步中输出的值逐渐接近目标…
阅读更多...
有问必答！zabbix“专家坐诊”第259期问答

有问必答！zabbix“专家坐诊”第259期问答

问题一 Q&#xff1a;现在监控项4万多&#xff0c;调整到多少比较合理 zabbix7.03&#xff1f; A&#xff1a;慢慢往上调&#xff0c;没有标准。 问题二 Q&#xff1a;想问下大家&#xff0c;zabbix的监控项怎么不能自动清除&#xff0c;比如说这次监控是A监控项&#xff0c;下…
阅读更多...
[LeetCode] 315. 计算右侧小于当前元素的个数

[LeetCode] 315. 计算右侧小于当前元素的个数

题目描述&#xff1a; 给你一个整数数组 nums &#xff0c;按要求返回一个新数组 counts 。数组 counts 有该性质&#xff1a; counts[i] 的值是 nums[i] 右侧小于 nums[i] 的元素的数量。 题目链接&#xff1a; . - 力扣&#xff08;LeetCode&#xff09; 题目主要思路&a…
阅读更多...
最新文章

Copyright @ 2022~2023