Eclipse-hono MQTT适配器:无法处理消息

Question

我想在没有集装箱平台的情况下在本地的机器上安装hono。我成功地运行了以下服务和先决条件：

• AMQP经纪人(RabbitMQ)
• InfluxDb
• Hono MQTT适配器
• 霍诺服务
• Hono设备注册表

当MQTT适配器启动时，我将得到以下日志：

14:37:48.257 [vert.x-eventloop-thread-0] INFO  o.e.h.c.RequestResponseClientConfigProperties - loading credentials for [127.0.0.1] from [/media/data/dev/hono-config/mqtt-adapter.credentials]
14:37:48.257 [main] DEBUG o.e.h.a.m.i.Application$$EnhancerBySpringCGLIB$$e74ec218 - Waiting 20 seconds for application to start up
14:37:48.258 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connecting to AMQP 1.0 container [amqp://127.0.0.1:5672]
14:37:48.472 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - starting attempt [#1] to connect to server [127.0.0.1:5672]
14:37:48.472 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connecting to AMQP 1.0 container [amqp://127.0.0.1:5672]
14:37:48.473 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - starting attempt [#1] to connect to server [127.0.0.1:5672]
14:37:48.473 [vert.x-eventloop-thread-0] INFO  o.e.h.c.RequestResponseClientConfigProperties - loading credentials for [127.0.0.1] from [/media/data/dev/hono-config/mqtt-adapter.credentials]
14:37:48.473 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connecting to AMQP 1.0 container [amqp://127.0.0.1:5672]
14:37:48.473 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - starting attempt [#1] to connect to server [127.0.0.1:5672]
14:37:48.473 [vert.x-eventloop-thread-0] INFO  o.e.h.c.RequestResponseClientConfigProperties - loading credentials for [127.0.0.1] from [/media/data/dev/hono-config/mqtt-adapter.credentials]
14:37:48.474 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connecting to AMQP 1.0 container [amqp://127.0.0.1:5672]
14:37:48.474 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - starting attempt [#1] to connect to server [127.0.0.1:5672]
14:37:48.474 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connecting to AMQP 1.0 container [amqp://127.0.0.1:5672]
14:37:48.474 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - limiting size of inbound message payload to 8096 bytes
14:37:48.474 [vert.x-eventloop-thread-0] WARN  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - authentication of devices turned off
14:37:48.474 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - Vertx native support: false
14:37:48.476 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - Server uses secure standard port 8883
14:37:48.479 [vert.x-eventloop-thread-0] DEBUG o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - OpenSSL [available: false, supports KeyManagerFactory: false]
14:37:48.479 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - using JDK's default SSL engine
14:37:48.480 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - enabling secure protocol [TLSv1.2]
14:37:48.530 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - MQTT server running on 0.0.0.0:8883
14:37:48.530 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - MQTT server running on 0.0.0.0:11883
14:37:48.533 [main] INFO  o.e.h.adapter.mqtt.impl.Application - Started Application in 2.542 seconds (JVM running for 3.01)
14:37:48.608 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connected to AMQP 1.0 container [amqp://127.0.0.1:5672], opening connection ...
14:37:48.611 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connected to AMQP 1.0 container [amqp://127.0.0.1:5672], opening connection ...
14:37:48.612 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connected to AMQP 1.0 container [amqp://127.0.0.1:5672], opening connection ...
14:37:48.612 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connected to AMQP 1.0 container [amqp://127.0.0.1:5672], opening connection ...
14:37:48.614 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connection to container [rabbit@bob-HP-ZBook-15-G4] at [amqp://127.0.0.1:5672] open
14:37:48.615 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connected to Tenant service
14:37:48.616 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connection to container [rabbit@bob-HP-ZBook-15-G4] at [amqp://127.0.0.1:5672] open
14:37:48.616 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connected to Messaging
14:37:48.616 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connection to container [rabbit@bob-HP-ZBook-15-G4] at [amqp://127.0.0.1:5672] open
14:37:48.616 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connected to Device Registration service
14:37:48.617 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connection to container [rabbit@bob-HP-ZBook-15-G4] at [amqp://127.0.0.1:5672] open
14:37:48.617 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connected to Credentials service
14:37:48.617 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connected to AMQP 1.0 container [amqp://127.0.0.1:5672], opening connection ...
14:37:48.618 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.ConnectionFactoryImpl - connection to container [rabbit@bob-HP-ZBook-15-G4] at [amqp://127.0.0.1:5672] open
14:37:48.618 [vert.x-eventloop-thread-0] INFO  o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connected to Command and Control service

当向MQTT适配器发送消息(mosquitto_pub -p 11883 -t遥测/DEFAULT_-t/4711 -m '{"temp"：5}')时(示例取自https://www.eclipse.org/hono/user-guide/mqtt-adapter/)，我将在MQTT适配器服务中获得以下日志记录：

14:38:05.208 [vert.x-eventloop-thread-0] DEBUG o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - connection request from client [client-id: mosqpub|21426-bob-HP-ZB]
14:38:05.213 [vert.x-eventloop-thread-0] DEBUG o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - unauthenticated device [clientId: mosqpub|21426-bob-HP-ZB] connected
14:38:05.215 [vert.x-eventloop-thread-0] INFO  o.e.h.s.m.LoggingConnectionEventProducer -    Connected - ID: mosqpub|21426-bob-HP-ZB, Protocol Adapter: hono-mqtt, Device: null, Data: null
14:38:05.222 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - creating new message sender for telemetry/DEFAULT_TENANT
14:38:05.236 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - creating new client [target: registration/DEFAULT_TENANT]
14:38:05.236 [vert.x-eventloop-thread-0] DEBUG o.e.h.c.impl.RegistrationClientImpl - creating new registration client for [DEFAULT_TENANT]
14:38:05.246 [vert.x-eventloop-thread-0] INFO  o.e.h.c.i.AbstractRequestResponseClient - enabling caching of responses from registration/DEFAULT_TENANT
14:38:05.253 [vert.x-eventloop-thread-0] DEBUG o.e.hono.client.impl.HonoClientImpl - creating new client [target: tenant]
14:38:05.254 [vert.x-eventloop-thread-0] DEBUG o.e.h.client.impl.TenantClientImpl - creating new tenant client
14:38:05.254 [vert.x-eventloop-thread-0] INFO  o.e.h.c.i.AbstractRequestResponseClient - enabling caching of responses from tenant
14:38:05.258 [vert.x-eventloop-thread-0] INFO  o.e.hono.client.impl.HonoClientImpl - remote server [127.0.0.1:5672] closed connection with error condition: The connections default session closed unexpectedly: : io.vertx.core.impl.NoStackTraceThrowable: Error{condition=amqp:invalid-field, description='Attach rejected: {unknown_destination,"telemetry/DEFAULT_TENANT"}', info=null}
14:38:05.260 [vert.x-eventloop-thread-0] DEBUG o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - cannot process message [endpoint: telemetry] from device [tenantId: DEFAULT_TENANT, deviceId: 4711]
org.eclipse.hono.client.ServerErrorException: no connection to service
    at org.eclipse.hono.client.impl.HonoClientImpl.lambda$getOrCreateSender$16(HonoClientImpl.java:619)
    at org.eclipse.hono.client.impl.HonoClientImpl.failAllCreationRequests(HonoClientImpl.java:463)
    at org.eclipse.hono.client.impl.HonoClientImpl.clearState(HonoClientImpl.java:455)
    at org.eclipse.hono.client.impl.HonoClientImpl.handleConnectionLoss(HonoClientImpl.java:437)
    at org.eclipse.hono.client.impl.HonoClientImpl.onRemoteClose(HonoClientImpl.java:417)
    at org.eclipse.hono.client.impl.HonoClientImpl.lambda$connect$2(HonoClientImpl.java:372)
    at io.vertx.proton.impl.ProtonConnectionImpl.lambda$getDefaultSession$6(ProtonConnectionImpl.java:256)
    at io.vertx.proton.impl.ProtonSessionImpl.fireRemoteClose(ProtonSessionImpl.java:270)
    at io.vertx.proton.impl.ProtonTransport.handleSocketBuffer(ProtonTransport.java:125)
    at io.vertx.core.net.impl.NetSocketImpl$DataMessageHandler.handle(NetSocketImpl.java:384)
    at io.vertx.core.net.impl.NetSocketImpl.handleMessageReceived(NetSocketImpl.java:351)
    at io.vertx.core.net.impl.NetClientImpl$1.handleMessage(NetClientImpl.java:242)
    at io.vertx.core.net.impl.NetClientImpl$1.handleMessage(NetClientImpl.java:239)
    at io.vertx.core.net.impl.VertxHandler.lambda$channelRead$1(VertxHandler.java:146)
    at io.vertx.core.impl.ContextImpl.lambda$wrapTask$2(ContextImpl.java:337)
    at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:195)
    at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:144)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1359)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:935)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:141)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:645)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:886)
    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
    at java.base/java.lang.Thread.run(Thread.java:844)
14:38:05.261 [vert.x-eventloop-thread-0] DEBUG o.e.h.a.m.i.VertxBasedMqttProtocolAdapter - discarding message [topic: telemetry/DEFAULT_TENANT/4711] from device: no connection to service

它试图连接到哪一项服务？一个与主题无关的小问题:在docker_swarm.sh中，我也注意到了消息传递服务，它用于什么？在https://www.eclipse.org/hono/getting-started/#starting-hono上，没有提到这一点。

提前感谢！

Answer 1

Hono不(正式)支持RabbitMQ作为AMQP1.0消息传递网络。但是，基于兔子对AMQP1.0的支持，您仍然可以让它工作。也就是说，MQTT适配器似乎无法打开到兔子的发送方链接，而兔子似乎拒绝了适配器在目标地址telemetry/DEFAULT_TENANT上打开发件人的请求。这可能是由于兔子作为消息代理的起源，它需要定义队列和/或主题，然后才能在运行时使用。

我不是RabbitMQ方面的专家，但我认为你应该弄清楚

• 如何配置RabbitMQ，以便当对等方打开相应的链接时，它在运行时动态创建队列/主题，或
• 在RabbitMQ前端定义相应的队列，例如通过将它们添加到兔子的配置文件中。

如果使用RabbitMQ并不是必需的，我实际上建议将Qpid调度路由器与Artemis一起使用。您将从Hono团队获得更好的支持，因为这是用于开发的AMQP消息传递网络的正式实现。

关于您的备注:在Hono 0.7之前，所有协议适配器都必须连接到消息传递服务，该服务在消息流中提供了一个中心点，在此消息流中，Hono可以在下游将其转发到AMQP消息传递网络(因此是消费应用程序)之前验证遥测消息或事件的起源。

为此，协议适配器需要在每条消息中包含一个JSON令牌，它从连接的设备的设备注册服务中获得JSON令牌，并声明该设备的注册状态。然后，Hono消息传递将验证JWT的签名，从而确保消息来源于实际启用协议适配器的租户的现有(注册)设备。这样做的想法是，设备注册表不会为适配器未授权的设备发出JWT。这对于自定义(第三方提供)协议适配器应与Hono安装集成(例如针对特定租户)的情况尤其有用，但应防止该协议适配器代表下游发送消息或不属于适配器授权的租户的任意设备。

在最近的版本中，我们放弃了这个组件，而倾向于在AMQP消息传递网络上授权协议适配器。

日志记录

所有Hono服务日志信息标准输出。可以通过Java属性来设置日志级别。如果没有设置，服务将记录在信息级别，只提供关于组件正在经历的生命周期事件的非常有限的信息。为了打开更详尽的调试日志记录，在启动JVM时设置类似于这个-Dspring.profiles.active=dev的系统属性。