java - 如何使用 HiveMq-mqtt 库连接到 Mqtt3AsyncClient？

Question

我一直在尝试在我的 Android 应用程序中使用 hivemq 实现 mqtt。尽管我使用了他们文档中的相同规范和配置，但我仍然无法建立成功的连接。

我之前能够将 paho 用于 mqtt，但如果应用程序处于后台，则它不适用于 android Oreo 及更高版本，因为该库没有 startServiceForeground 更新。所以搬到了HiveMq。我只是使用此处的快速入门指南试用他们的示例应用程序。1. https://hivemq.github.io/hivemq-mqtt-client/2. https://www.hivemq.com/blog/mqtt-client-library-enyclopedia-hivemq-mqtt-client/3. https://github.com/hivemq/hivemq-mqtt-client/blob/develop/README.md几乎浪费了一天的时间来研究如何连接到他们的 mqtt，但没有成功

lateinit var client : Mqtt3Client

    fun connect() {
        client = Mqtt3Client.builder()
             .identifier(UUID.randomUUID().toString())
            .serverHost("broker.hivemq.com")
            .serverPort(1883)
            .buildAsync()


            client.toAsync().connect()
            .whenComplete { mqtt3ConnAck, throwable ->
                if (throwable != null) {
                    // handle failure
                    android.util.Log.v("HIVE-MQTT-LCDP", " connection failed")
                } else {
                    android.util.Log.v("HIVE-MQTT-LCDP", " connected")
                    // setup subscribes or start publishing
                    subscribe()
                    publish()
                }
            }

        fab.setOnLongClickListener {
            android.util.Log.v("HIVE-MQTT-LCDP", " disconnected")
            client.toAsync().disconnect()
            true
        }

    }


    fun publish() {
        client.toAsync().publishWith()
            .topic("the/topic")
            .payload("payload".toByteArray())
            .send()
            .whenComplete { mqtt3Publish, throwable ->
                if (throwable != null) {
                    android.util.Log.v("HIVE-MQTT-LCDP", " failure to publish")
                    // handle failure to publish
                } else {
                    android.util.Log.v("HIVE-MQTT-LCDP", " successful to publish")
                    // handle successful publish, e.g. logging or incrementing a metric
                }
            }
    }

    fun subscribe() {
        client.toAsync().subscribeWith()
            .topicFilter("the/topic")
            .callback { mqtt3Publish ->
                // Process the received message
                android.util.Log.v("HIVE-MQTT-LCDP", " Message received")
            }
            .send()
            .whenComplete { mqtt3SubAck, throwable ->
                if (throwable != null) {
                    android.util.Log.v("HIVE-MQTT-LCDP", " failure to subscribe")
                    // Handle failure to subscribe
                } else {
                    android.util.Log.v("HIVE-MQTT-LCDP", " successful to subscribe")
                    // Handle successful subscription, e.g. logging or incrementing a metric
                }
            }
    }```

expecting a response in the subscribe-whenComplete.

the clues:

it throws the exception "com.hivemq.client.mqtt.exceptions.ConnectionClosedException: Server closed connection without DISCONNECT." in the addConnectionListener

Stacktrace:

   Caused by: java.lang.ClassNotFoundException: Didn't find class "org.apache.logging.log4j.spi.ExtendedLoggerWrapper" on path: DexPathList[[zip file "/data/app/com.example.myapplication-jscQ0Rz9LiCQTlaR1v-Z5w==/base.apk"],nativeLibraryDirectories=[/data/app/com.example.myapplication-jscQ0Rz9LiCQTlaR1v-Z5w==/lib/arm64, /system/lib64, /system/vendor/lib64]]
        at java.lang.Class dalvik.system.BaseDexClassLoader.findClass(java.lang.String) (BaseDexClassLoader.java:93)
        at java.lang.Class java.lang.ClassLoader.loadClass(java.lang.String, boolean) (ClassLoader.java:379)
        at java.lang.Class java.lang.ClassLoader.loadClass(java.lang.String) (ClassLoader.java:312)
        at io.netty.util.internal.logging.InternalLogger io.netty.util.internal.logging.Log4J2LoggerFactory.newInstance(java.lang.String) (Log4J2LoggerFactory.java:33)
        at io.netty.util.internal.logging.InternalLoggerFactory io.netty.util.internal.logging.InternalLoggerFactory.newDefaultFactory(java.lang.String) (InternalLoggerFactory.java:51)
        at io.netty.util.internal.logging.InternalLoggerFactory io.netty.util.internal.logging.InternalLoggerFactory.getDefaultFactory() (InternalLoggerFactory.java:67)
        at io.netty.util.internal.logging.InternalLogger io.netty.util.internal.logging.InternalLoggerFactory.getInstance(java.lang.String) (InternalLoggerFactory.java:93)
        at io.netty.util.internal.logging.InternalLogger io.netty.util.internal.logging.InternalLoggerFactory.getInstance(java.lang.Class) (InternalLoggerFactory.java:86)
        at void io.netty.util.internal.SystemPropertyUtil.<clinit>() (SystemPropertyUtil.java:29)
        at boolean io.netty.util.internal.SystemPropertyUtil.getBoolean(java.lang.String, boolean) (SystemPropertyUtil.java:99)
        at void io.netty.channel.epoll.Epoll.<clinit>() (Epoll.java:31)
        at boolean io.netty.channel.epoll.Epoll.isAvailable() (Epoll.java:68)
        at com.hivemq.client.internal.mqtt.netty.NettyEventLoopProvider com.hivemq.client.internal.mqtt.netty.NettyModule.provideNettyEventLoopProvider() (NettyModule.java:40)
        at com.hivemq.client.internal.mqtt.netty.NettyEventLoopProvider com.hivemq.client.internal.mqtt.netty.NettyModule_ProvideNettyEventLoopProviderFactory.proxyProvideNettyEventLoopProvider() (NettyModule_ProvideNettyEventLoopProviderFactory.java:31)
        at com.hivemq.client.internal.mqtt.netty.NettyEventLoopProvider com.hivemq.client.internal.mqtt.netty.NettyModule_ProvideNettyEventLoopProviderFactory.provideInstance() (NettyModule_ProvideNettyEventLoopProviderFactory.java:22)
        at com.hivemq.client.internal.mqtt.netty.NettyEventLoopProvider com.hivemq.client.internal.mqtt.netty.NettyModule_ProvideNettyEventLoopProviderFactory.get() (NettyModule_ProvideNettyEventLoopProviderFactory.java:18)
        at java.lang.Object com.hivemq.client.internal.mqtt.netty.NettyModule_ProvideNettyEventLoopProviderFactory.get() (NettyModule_ProvideNettyEventLoopProviderFactory.java:7)
        at java.lang.Object dagger.internal.DoubleCheck.get() (DoubleCheck.java:47)
        at com.hivemq.client.internal.mqtt.netty.NettyEventLoopProvider com.hivemq.client.internal.mqtt.ioc.DaggerSingletonComponent.nettyEventLoopProvider() (DaggerSingletonComponent.java:377)
        at io.netty.channel.EventLoop com.hivemq.client.internal.mqtt.MqttClientConfig.acquireEventLoop() (MqttClientConfig.java:174)
        at void com.hivemq.client.internal.mqtt.handler.connect.MqttConnAckSingle.subscribeActual(io.reactivex.SingleObserver) (MqttConnAckSingle.java:69)
        at void io.reactivex.Single.subscribe(io.reactivex.SingleObserver) (Single.java:3575)
        at void io.reactivex.internal.operators.single.SingleObserveOn.subscribeActual(io.reactivex.SingleObserver) (SingleObserveOn.java:35)
        at void io.reactivex.Single.subscribe(io.reactivex.SingleObserver) (Single.java:3575)
        at void com.hivemq.client.internal.rx.RxFutureConverter$RxJavaSingleFuture.<init>(io.reactivex.Single) (RxFutureConverter.java:114)
I/zygote64:     at java.util.concurrent.CompletableFuture com.hivemq.client.internal.rx.RxFutureConverter.toFuture(io.reactivex.Single) (RxFutureConverter.java:44)
        at java.util.concurrent.CompletableFuture com.hivemq.client.internal.mqtt.MqttAsyncClient.connect(com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5Connect) (MqttAsyncClient.java:70)
        at java.util.concurrent.CompletableFuture com.hivemq.client.internal.mqtt.mqtt3.Mqtt3AsyncClientView.connect(com.hivemq.client.mqtt.mqtt3.message.connect.Mqtt3Connect) (Mqtt3AsyncClientView.java:116)
        at java.util.concurrent.CompletableFuture com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient.connect() (Mqtt3AsyncClient.java:59)
        at void com.example.myapplication.MainActivity.connect() (MainActivity.kt:70)
        at void com.example.myapplication.MainActivity.onCreate(android.os.Bundle) (MainActivity.kt:58)
        at void android.app.Activity.performCreate(android.os.Bundle) (Activity.java:7183)
        at void android.app.Instrumentation.callActivityOnCreate(android.app.Activity, android.os.Bundle) (Instrumentation.java:1220)
        at android.app.Activity android.app.ActivityThread.performLaunchActivity(android.app.ActivityThread$ActivityClientRecord, android.content.Intent) (ActivityThread.java:2910)
        at void android.app.ActivityThread.handleLaunchActivity(android.app.ActivityThread$ActivityClientRecord, android.content.Intent, java.lang.String) (ActivityThread.java:3032)
        at void android.app.ActivityThread.-wrap11(android.app.ActivityThread, android.app.ActivityThread$ActivityClientRecord, android.content.Intent, java.lang.String) (ActivityThread.java:-1)
        at void android.app.ActivityThread$H.handleMessage(android.os.Message) (ActivityThread.java:1696)
        at void android.os.Handler.dispatchMessage(android.os.Message) (Handler.java:105)
        at void android.os.Looper.loop() (Looper.java:164)
        at void android.app.ActivityThread.main(java.lang.String[]) (ActivityThread.java:6944)
        at java.lang.Object java.lang.reflect.Method.invoke(java.lang.Object, java.lang.Object[]) (Method.java:-2)
        at void com.android.internal.os.Zygote$MethodAndArgsCaller.run() (Zygote.java:327)
        at void com.android.internal.os.ZygoteInit.main(java.lang.String[]) (ZygoteInit.java:1374)
E/zygote64: No implementation found for int io.netty.channel.epoll.Native.offsetofEpollData() (tried Java_io_netty_channel_epoll_Native_offsetofEpollData and Java_io_netty_channel_epoll_Native_offsetofEpollData__)
I/PlatformDependent: Your platform does not provide complete low-level API for accessing direct buffers reliably. Unless explicitly requested, heap buffer will always be preferred to avoid potential system instability.
W/io.netty.util.NetUtil: Failed to find the loopback interface
W/MacAddressUtil: Failed to find a usable hardware address from the network interfaces; using random bytes: 20:c2:d8:6a:3f:bc:f6:90
W/DefaultPromise: An exception was thrown by com.hivemq.client.internal.mqtt.handler.connect.-$$Lambda$MqttConnAckSingle$1NP8pOpp3tvaWFx_4_jMw-CQFiU.operationComplete()
    java.lang.IllegalStateException: MqttClientReconnector must be called from the eventLoop.
        at com.hivemq.client.internal.mqtt.lifecycle.MqttClientReconnector.checkThread(MqttClientReconnector.java:158)
        at com.hivemq.client.internal.mqtt.lifecycle.MqttClientReconnector.isReconnect(MqttClientReconnector.java:93)
        at com.hivemq.client.internal.mqtt.handler.connect.MqttConnAckSingle.reconnect(MqttConnAckSingle.java:135)
        at com.hivemq.client.internal.mqtt.handler.connect.MqttConnAckSingle.reconnect(MqttConnAckSingle.java:103)
        at com.hivemq.client.internal.mqtt.handler.connect.MqttConnAckSingle.lambda$connect$0(MqttConnAckSingle.java:90)
        at com.hivemq.client.internal.mqtt.handler.connect.-$$Lambda$MqttConnAckSingle$1NP8pOpp3tvaWFx_4_jMw-CQFiU.operationComplete(Unknown Source:8)
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:511)
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:485)
        at io.netty.util.concurrent.DefaultPromise.access$000(DefaultPromise.java:33)
        at io.netty.util.concurrent.DefaultPromise$1.run(DefaultPromise.java:435)
        at io.netty.util.concurrent.GlobalEventExecutor$TaskRunner.run(GlobalEventExecutor.java:248)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.lang.Thread.run(Thread.java:764)
D/OpenGLRenderer: HWUI GL Pipeline
D/ViewRootImpl@2daeee6[MainActivity]: setView = DecorView@cea6ed4[MainActivity] TM=true MM=false
V/InputMethodManager: Not IME target window, ignoring
D/ViewRootImpl@2daeee6[MainActivity]: dispatchAttachedToWindow
V/Surface: sf_framedrop debug : 0x4f4c, game : false, logging : 0
D/ViewRootImpl@2daeee6[MainActivity]: Relayout returned: old=[0,0][0,0] new=[0,0][1080,1920] result=0x7 surface={valid=true 481715916800} changed=true
I/OpenGLRenderer: Initialized EGL, version 1.4
D/OpenGLRenderer: Swap behavior 2
D/libGLESv1: STS_GLApi : DTS, ODTC are not allowed for Package : com.example.myapplication
D/mali_winsys: EGLint new_window_surface(egl_winsys_display *, void *, EGLSurface, EGLConfig, egl_winsys_surface **, egl_color_buffer_format *, EGLBoolean) returns 0x3000,  [1080x1920]-format:1
D/OpenGLRenderer: eglCreateWindowSurface = 0x702882f2b0
E/System: Uncaught exception thrown by finalizer
E/System: java.lang.NullPointerException: Attempt to invoke virtual method 'boolean java.io.FileDescriptor.isSocket$()' on a null object reference
        at sun.nio.ch.FileDispatcherImpl.preCloseImpl(FileDispatcherImpl.java:115)
        at sun.nio.ch.SocketDispatcher.preClose(SocketDispatcher.java:66)
        at sun.nio.ch.SocketChannelImpl.implCloseSelectableChannel(SocketChannelImpl.java:883)
        at java.nio.channels.spi.AbstractSelectableChannel.implCloseChannel(AbstractSelectableChannel.java:234)
        at java.nio.channels.spi.AbstractInterruptibleChannel.close(AbstractInterruptibleChannel.java:116)
        at sun.nio.ch.SocketChannelImpl.finalize(SocketChannelImpl.java:937)
        at java.lang.Daemons$FinalizerDaemon.doFinalize(Daemons.java:250)
        at java.lang.Daemons$FinalizerDaemon.runInternal(Daemons.java:237)
        at java.lang.Daemons$Daemon.run(Daemons.java:103)
        at java.lang.Thread.run(Thread.java:764)
I/zygote64: Do partial code cache collection, code=61KB, data=53KB
I/zygote64: After code cache collection, code=61KB, data=53KB
    Increasing code cache capacity to 256KB
D/ViewRootImpl@2daeee6[MainActivity]: Relayout returned: old=[0,0][1080,1920] new=[0,0][1080,1920] result=0x3 surface={valid=true 481715916800} changed=false
I/zygote64: Compiler allocated 9MB to compile void android.widget.TextView.<init>(android.content.Context, android.util.AttributeSet, int, int)
D/ViewRootImpl@2daeee6[MainActivity]: MSG_RESIZED_REPORT: frame=Rect(0, 0 - 1080, 1920) ci=Rect(0, 72 - 0, 0) vi=Rect(0, 72 - 0, 0) or=1
D/ViewRootImpl@2daeee6[MainActivity]: MSG_WINDOW_FOCUS_CHANGED 1
V/InputMethodManager: Starting input: tba=android.view.inputmethod.EditorInfo@159ea78 nm : com.example.myapplication ic=null
I/InputMethodManager: startInputInner - mService.startInputOrWindowGainedFocus
V/InputMethodManager: Starting input: tba=android.view.inputmethod.EditorInfo@9899551 nm : com.example.myapplication ic=null
D/ViewRootImpl@2daeee6[MainActivity]: ViewPostIme pointer 0
D/ViewRootImpl@2daeee6[MainActivity]: ViewPostIme pointer 1
V/HIVE-MQTT-LCDP:  failure to publish
I/zygote64: Do full code cache collection, code=123KB, data=95KB
I/zygote64: After code cache collection, code=102KB, data=64KB

Answer 1

您必须将以下权限添加到您的 AndroidManifest.xml 中:

<uses-permission android:name="android.permission.INTERNET"/>

(参见 https://github.com/hivemq/hivemq-mqtt-client/issues/213)

ClassNotFoundException 只是一条日志消息，不是错误。 Netty(HiveMQ MQTT 客户端使用的网络库)尝试检测您是否加载了日志记录框架。因此，添加日志框架是可选的。

---

Proguard 更新:

Netty 和 JCTools 不能很好地保护，所以添加这些保护规则:

-keepclassmembernames class io.netty.** {
    *;
}

-keepclassmembernames class org.jctools.** {
    *;
}