问题标签 [spring-integration-mqtt]

这就是我使用spring集成定义我的mqtt连接的方式。我不确定这是否可能bt我们可以在收到10条消息后设置一个mqtt订阅者工作吗？现在订阅者在发布消息后正常工作。

我正在研究使用 Apache Camel（使用 camel-mqtt）+ Spring Boot 构建 AWS IoT Java 客户端。这听起来很适合我，但找不到任何例子。有什么我看不到的缺点吗？有兴趣看到任何指针。

我已经mqttlens从google chorme apps.

它在几周内运行良好，但最近当我启动它时它出现在缩放中。

我该如何解决？

.

我正在使用 Spring 的 spring-integration-mqtt，我可以连接到单个 Mqtt 服务器并可以接收有关订阅主题的消息，现在我想制作可以连接到多个 Mqtt 服务器并可以从每个连接接收数据的应用程序，我想要将其作为动态管理，我可以从数据库或文本文件中添加更多 Mqtt 服务器。

用于订阅的单个 Mqtt 连接的简单 bean 如下

上面的代码为 mqtt 服务器创建了一个连接，并且可以接收消息，如果我将相同的代码复制两次粘贴到具有不同 Mqtt ip 地址的第二个服务器，我可以连接到两个 Mqtt 服务器，如下所示

上面的代码也可以正常工作，我可以从两个 Mqtt 服务器接收消息，但是有什么方法可以动态管理它，如下所示，我将 bean 的返回类型更改为列表，但没有工作：

有什么方法可以动态管理这些bean，我可以从文本文件和for循环中获取mqtt服务器详细信息，或者我可以管理多个连接。

在 Spring 集成应用程序中，我使用并发消费者一次消费和处理多条消息。在我的应用程序中，我将所有 bean 配置为一个单例。我假设我是否要通过使用并发消费者的并行化处理，多条消息输入到相同的集成组件中。它会导致两个对象之间的数据冲突吗？

我有这段代码在 Spring Integration 5.0.x / Boot 2.0 下接收 MQTT 消息。它适用于文本消息，但是当我尝试处理二进制消息时它会失败，因为发生转换String并且这会破坏内容（在这种情况下：png 图像文件）。

我怎样才能收到未篡改的消息？

我试着setBytesMessageMapperon DefaultPahoMessageConverter，但这并没有改变任何事情。当我使用 mqtt.fx 下载消息内容时，我可以证明二进制内容设置正确，所以我确信这是接收端的问题。

我尝试测试使用 Spring Boot 和 Spring Integration 创建的简单 MQTT 侦听器，但无法正常工作。我尝试了很多方法。最有希望的是：

但是，测试运行时 2018-07-12 16:53:50.937 ERROR 21160 --- [T Rec: consumer] .m.i.MqttPahoMessageDrivenChannelAdapter : Lost connection: Verbindung wurde getrennt; retrying... ，我没有看到任何打印出来的东西。代码主要来自：https ://github.com/spring-projects/spring-integration-samples/tree/master/basic/mqtt 。该示例运行良好，但我需要能够编写适当的集成测试。

配置是：

更新：

也没有用：

更新

更简单......同样的错误：

我是 Spring-Integration 的新手，正在寻找以下任务的解决方案。

我们尝试创建一个 MQTT-Consumer（基于mqtt 入站通道适配器），它接收来自多达 50 个发布者的 7 个主题的消息。

由于适配器按顺序处理消息，因此必须扩展并行处理，或者获取代理丢弃的其余消息。

扩展的第一个想法是每个主题有一个适配器，并带有相应的转换器。它可以很好地消耗来自一个发布者的消息，但不能承受如此多的发布者，使用排队或直接出站通道。

目前正在考虑将执行程序通道与每个主题的消息处理程序（自己的转换器）的数量一起使用。之后的那些将他们的结果传送到相应的排队通道中，这将允许批量处理和传播消息。

但是，我不确定为每个主题创建单个适配器是否是一个好主意，并且正在寻找有关架构改进的任何建议。提前致谢！

我在 spring 中使用 spring mqtt 集成配置了一个 MQTT 订阅者。在 handleMessage 方法中，我正在执行某些需要时间的业务逻辑。在测试时，我注意到当我发送大量消息时，代理重新发布与原始消息相同的消息（我检查了消息有效负载是否重复，它是作为原始消息发送的）。MQTT 代理甚至在订阅者可以发送 PUCOMP 之前再次发布消息。QOS 级别设置为 2

我正在尝试学习如何在Spring-Integration中处理MQTT 消息。已创建一个转换器，它为每个 MQTT 主题订阅一个 MqttPahoMessageDrivenChannelAdapter 以使用和转换消息。

问题是我们的数据提供者正计划“加速”发布消息。因此，不是有几个（<=10）主题，每个主题都有大约 150 个字段的消息，而是计划将这些字段中的每一个发布到单独的 MQTT 主题。

这意味着我的转换器将不得不消耗 ca。1000个mqtt主题，但不知道是否：

1. spring-integration 是否仍然是一个不错的选择。引起afaik。提到的适配器使用 PAHO MqttClient，它将在一个线程中使用它订阅的所有主题的消息，并且创建这些适配器的 1000 个实例是一种过度杀伤力。
2. 如果我们进一步坚持使用 spring-integration 并使用提供的组件，那么为所有字段创建一个单一的入站适配器是否是一个好主意，这些字段以前在一个主题的消息中，但将转换从适配器 bean 转移到一个单独的 bean（进行转换）通过执行器通道连接到适配器，从而在某个线程池上并行执行这些字段的转换。

提前感谢您的回答！