Spring 集成 MQTT cleanSession=false 失败，因为客户端取消订阅主题
Spring Integration MQTT cleanSession=false fails because client unsubscribes from topics
我正在配置 Spring 集成以在我的一个频道上使用 cleanSession=false。
<bean id="clientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="cleanSession" value="false" />
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttLiveDataInbound"
client-id="client1"
url="${mqtt.broker.url}"
qos="1"
topics="liveData"
client-factory="clientFactory"
channel="channelLiveData"/>
原因是我希望能够在我的应用程序离线时接收消息。当我的应用程序重新启动时，我希望它接收在我离开期间发布的任何 QoS>0 消息。
现在我注意到一些奇怪的事情：我的应用程序在停机后没有接收到丢失的 QoS>0 消息。
我记录了一个简单的场景，其中
- Spring 集成开始
- QoS1 消息发送到主题并由 Spring 集成接收
- Spring 集成退出
- 向主题发送 QoS1 消息
- Spring 集成开始
- Spring 集成未收到离线时发送的 QoS1 消息。
原因如下（从下面的日志中可以看出）：
- spring 集成退出
- 取消订阅主题
- 它会断开客户端连接
这实际上是在告诉代理这个客户不再对这些消息感兴趣。当我的应用程序关闭时，代理不再为我保留这些 QoS>0 消息。
当我的应用程序再次启动时，它无法接收在它关闭时发布的 QoS>0 消息。
1448917620: New connection from 127.0.0.1 on port 1883.
1448917620: New client connected from 127.0.0.1 as client1 (c0, k60).
1448917620: Sending CONNACK to client1 (0, 0)
1448917620: Received SUBSCRIBE from client1
1448917620: liveData (QoS 1)
1448917620: Sending SUBACK to client1
1448917632: New connection from ::1 on port 1883.
1448917632: New client connected from ::1 as mosqpub/25936-MacBook-P (c1, k60, u'system').
1448917632: Sending CONNACK to mosqpub/25936-MacBook-P (0, 0)
1448917632: Received PUBLISH from mosqpub/25936-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917632: Sending PUBACK to mosqpub/25936-MacBook-P (Mid: 1)
1448917632: Sending PUBLISH to client1 (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917632: Received DISCONNECT from mosqpub/25936-MacBook-P
1448917632: Client mosqpub/25936-MacBook-P disconnected.
1448917633: Received PUBACK from client1 (Mid: 1)
1448917643: Received UNSUBSCRIBE from client1
1448917643: liveData
1448917643: Received DISCONNECT from client1
1448917643: Client client1 disconnected.
1448917648: New connection from ::1 on port 1883.
1448917648: New client connected from ::1 as mosqpub/25945-MacBook-P (c1, k60, u'system').
1448917648: Sending CONNACK to mosqpub/25945-MacBook-P (0, 0)
1448917648: Received PUBLISH from mosqpub/25945-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917648: Sending PUBACK to mosqpub/25945-MacBook-P (Mid: 1)
1448917648: Received DISCONNECT from mosqpub/25945-MacBook-P
1448917648: Client mosqpub/25945-MacBook-P disconnected.
1448917665: New connection from 127.0.0.1 on port 1883.
1448917665: Client client1 disconnected.
1448917665: New client connected from 127.0.0.1 as client1 (c0, k60).
1448917665: Sending CONNACK to client1 (1, 0)
1448917665: Received SUBSCRIBE from client1
1448917665: liveData (QoS 1)
1448917665: Sending SUBACK to client1
我运行这个场景使用 mosquitto 客户端工具，退出 mosquitto 订阅者会断开客户端连接，但不会取消订阅主题
1448917534: New connection from ::1 on port 1883.
1448917534: New client connected from ::1 as client1 (c0, k60).
1448917534: Sending CONNACK to client1 (0, 0)
1448917534: Received SUBSCRIBE from client1
1448917534: liveData (QoS 1)
1448917534: Sending SUBACK to client1
1448917550: New connection from ::1 on port 1883.
1448917550: New client connected from ::1 as mosqpub/25879-MacBook-P (c1, k60, u'system').
1448917550: Sending CONNACK to mosqpub/25879-MacBook-P (0, 0)
1448917550: Received PUBLISH from mosqpub/25879-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917550: Sending PUBACK to mosqpub/25879-MacBook-P (Mid: 1)
1448917550: Sending PUBLISH to client1 (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917550: Received DISCONNECT from mosqpub/25879-MacBook-P
1448917550: Client mosqpub/25879-MacBook-P disconnected.
1448917550: Received PUBACK from client1 (Mid: 1)
1448917553: Socket error on client client1, disconnecting.
1448917554: New connection from ::1 on port 1883.
1448917554: New client connected from ::1 as mosqpub/25884-MacBook-P (c1, k60, u'system').
1448917554: Sending CONNACK to mosqpub/25884-MacBook-P (0, 0)
1448917554: Received PUBLISH from mosqpub/25884-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917554: Sending PUBACK to mosqpub/25884-MacBook-P (Mid: 1)
1448917554: Received DISCONNECT from mosqpub/25884-MacBook-P
1448917555: Client mosqpub/25884-MacBook-P disconnected.
1448917556: New connection from ::1 on port 1883.
1448917556: Client client1 disconnected.
1448917556: New client connected from ::1 as client1 (c0, k60).
1448917556: Sending CONNACK to client1 (0, 0)
1448917556: Sending PUBLISH to client1 (d0, q1, r0, m2, 'liveData', ... (68 bytes))
1448917556: Received SUBSCRIBE from client1
1448917556: liveData (QoS 1)
1448917556: Sending SUBACK to client1
1448917556: Received PUBACK from client1 (Mid: 2)
知道如何处理这种情况吗？
编辑：
按照接受的答案中提出的解决方法实施时，出现以下错误。我的 Spring 上下文是从网络应用加载的。我试过将 IgnoreUnsubscribePahoClientFactory 放在一个单独的 JAR（与 spring-integration / paho 相同级别）以及 webapp 类 本身中。
2015-12-02 15:47:43,703 ERROR org.springframework.integration.handler.LoggingHandler - org.springframework.aop.framework.AopConfigException: Could not generate CGLIB subclass of class [class org.eclipse.paho.client.mqttv3.MqttAsyncClient]: Common causes of this problem include using a final class or a non-visible class; nested exception is org.springframework.cglib.core.CodeGenerationException: java.lang.reflect.InvocationTargetException-->null
at org.springframework.aop.framework.CglibAopProxy.getProxy(CglibAopProxy.java:206)
at org.springframework.aop.framework.ProxyFactoryBean.getProxy(ProxyFactoryBean.java:368)
at org.springframework.aop.framework.ProxyFactoryBean.getSingletonInstance(ProxyFactoryBean.java:322)
at org.springframework.aop.framework.ProxyFactoryBean.getObject(ProxyFactoryBean.java:246)
at com.ecs.vdm.rest.integration.IgnoreUnsubscribePahoClientFactory.proxy(IgnoreUnsubscribePahoClientFactory.java:62)
at com.ecs.vdm.rest.integration.IgnoreUnsubscribePahoClientFactory.getAsyncClientInstance(IgnoreUnsubscribePahoClientFactory.java:43)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.connectAndSubscribe(MqttPahoMessageDrivenChannelAdapter.java:216)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.access0(MqttPahoMessageDrivenChannelAdapter.java:45)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.run(MqttPahoMessageDrivenChannelAdapter.java:272)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:178)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.cglib.core.CodeGenerationException: java.lang.reflect.InvocationTargetException-->null
at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:237)
at org.springframework.cglib.proxy.Enhancer.createHelper(Enhancer.java:377)
at org.springframework.cglib.proxy.Enhancer.createClass(Enhancer.java:317)
at org.springframework.aop.framework.ObjenesisCglibAopProxy.createProxyClassAndInstance(ObjenesisCglibAopProxy.java:57)
at org.springframework.aop.framework.CglibAopProxy.getProxy(CglibAopProxy.java:202)
... 16 more
Caused by: java.lang.reflect.InvocationTargetException
at sun.reflect.GeneratedMethodAccessor37.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.springframework.cglib.core.ReflectUtils.defineClass(ReflectUtils.java:384)
at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:219)
... 20 more
Caused by: java.lang.SecurityException: class "org.eclipse.paho.client.mqttv3.MqttAsyncClient$$EnhancerBySpringCGLIB$$d14754a9_4603"'s signer information does not match signer information of other classes in the same package
at java.lang.ClassLoader.checkCerts(ClassLoader.java:952)
at java.lang.ClassLoader.preDefineClass(ClassLoader.java:666)
at java.lang.ClassLoader.defineClass(ClassLoader.java:794)
... 25 more
这是一个错误 - 它在 stop() 期间无条件取消订阅。
我没有看到简单的解决方法；但我有几个想法；我会post这里if/when我有事
同时，请打开一个JIRA Issue。
编辑
这有点像大锤，但它应该适合你；它有效地忽略了客户端上对 unsubscribe 的调用。它可以做得更复杂一些，只在 QOS > 0 时忽略，但这会涉及更多。
如果您已经在使用 DefaultMqttPahoClientFactory，只需将 bean class 更改为这个。如果您当前未使用工厂，请将其声明为 bean 并使用 client-factory 属性将其提供给适配器。
我们会在即将发布的版本中妥善修复它。
我正在配置 Spring 集成以在我的一个频道上使用 cleanSession=false。
<bean id="clientFactory" class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
<property name="cleanSession" value="false" />
</bean>
<int-mqtt:message-driven-channel-adapter id="mqttLiveDataInbound"
client-id="client1"
url="${mqtt.broker.url}"
qos="1"
topics="liveData"
client-factory="clientFactory"
channel="channelLiveData"/>
原因是我希望能够在我的应用程序离线时接收消息。当我的应用程序重新启动时，我希望它接收在我离开期间发布的任何 QoS>0 消息。
现在我注意到一些奇怪的事情：我的应用程序在停机后没有接收到丢失的 QoS>0 消息。
我记录了一个简单的场景，其中
- Spring 集成开始
- QoS1 消息发送到主题并由 Spring 集成接收
- Spring 集成退出
- 向主题发送 QoS1 消息
- Spring 集成开始
- Spring 集成未收到离线时发送的 QoS1 消息。
原因如下（从下面的日志中可以看出）：
- spring 集成退出
- 取消订阅主题
- 它会断开客户端连接
这实际上是在告诉代理这个客户不再对这些消息感兴趣。当我的应用程序关闭时，代理不再为我保留这些 QoS>0 消息。
当我的应用程序再次启动时，它无法接收在它关闭时发布的 QoS>0 消息。
1448917620: New connection from 127.0.0.1 on port 1883.
1448917620: New client connected from 127.0.0.1 as client1 (c0, k60).
1448917620: Sending CONNACK to client1 (0, 0)
1448917620: Received SUBSCRIBE from client1
1448917620: liveData (QoS 1)
1448917620: Sending SUBACK to client1
1448917632: New connection from ::1 on port 1883.
1448917632: New client connected from ::1 as mosqpub/25936-MacBook-P (c1, k60, u'system').
1448917632: Sending CONNACK to mosqpub/25936-MacBook-P (0, 0)
1448917632: Received PUBLISH from mosqpub/25936-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917632: Sending PUBACK to mosqpub/25936-MacBook-P (Mid: 1)
1448917632: Sending PUBLISH to client1 (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917632: Received DISCONNECT from mosqpub/25936-MacBook-P
1448917632: Client mosqpub/25936-MacBook-P disconnected.
1448917633: Received PUBACK from client1 (Mid: 1)
1448917643: Received UNSUBSCRIBE from client1
1448917643: liveData
1448917643: Received DISCONNECT from client1
1448917643: Client client1 disconnected.
1448917648: New connection from ::1 on port 1883.
1448917648: New client connected from ::1 as mosqpub/25945-MacBook-P (c1, k60, u'system').
1448917648: Sending CONNACK to mosqpub/25945-MacBook-P (0, 0)
1448917648: Received PUBLISH from mosqpub/25945-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917648: Sending PUBACK to mosqpub/25945-MacBook-P (Mid: 1)
1448917648: Received DISCONNECT from mosqpub/25945-MacBook-P
1448917648: Client mosqpub/25945-MacBook-P disconnected.
1448917665: New connection from 127.0.0.1 on port 1883.
1448917665: Client client1 disconnected.
1448917665: New client connected from 127.0.0.1 as client1 (c0, k60).
1448917665: Sending CONNACK to client1 (1, 0)
1448917665: Received SUBSCRIBE from client1
1448917665: liveData (QoS 1)
1448917665: Sending SUBACK to client1
我运行这个场景使用 mosquitto 客户端工具，退出 mosquitto 订阅者会断开客户端连接，但不会取消订阅主题
1448917534: New connection from ::1 on port 1883.
1448917534: New client connected from ::1 as client1 (c0, k60).
1448917534: Sending CONNACK to client1 (0, 0)
1448917534: Received SUBSCRIBE from client1
1448917534: liveData (QoS 1)
1448917534: Sending SUBACK to client1
1448917550: New connection from ::1 on port 1883.
1448917550: New client connected from ::1 as mosqpub/25879-MacBook-P (c1, k60, u'system').
1448917550: Sending CONNACK to mosqpub/25879-MacBook-P (0, 0)
1448917550: Received PUBLISH from mosqpub/25879-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917550: Sending PUBACK to mosqpub/25879-MacBook-P (Mid: 1)
1448917550: Sending PUBLISH to client1 (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917550: Received DISCONNECT from mosqpub/25879-MacBook-P
1448917550: Client mosqpub/25879-MacBook-P disconnected.
1448917550: Received PUBACK from client1 (Mid: 1)
1448917553: Socket error on client client1, disconnecting.
1448917554: New connection from ::1 on port 1883.
1448917554: New client connected from ::1 as mosqpub/25884-MacBook-P (c1, k60, u'system').
1448917554: Sending CONNACK to mosqpub/25884-MacBook-P (0, 0)
1448917554: Received PUBLISH from mosqpub/25884-MacBook-P (d0, q1, r0, m1, 'liveData', ... (68 bytes))
1448917554: Sending PUBACK to mosqpub/25884-MacBook-P (Mid: 1)
1448917554: Received DISCONNECT from mosqpub/25884-MacBook-P
1448917555: Client mosqpub/25884-MacBook-P disconnected.
1448917556: New connection from ::1 on port 1883.
1448917556: Client client1 disconnected.
1448917556: New client connected from ::1 as client1 (c0, k60).
1448917556: Sending CONNACK to client1 (0, 0)
1448917556: Sending PUBLISH to client1 (d0, q1, r0, m2, 'liveData', ... (68 bytes))
1448917556: Received SUBSCRIBE from client1
1448917556: liveData (QoS 1)
1448917556: Sending SUBACK to client1
1448917556: Received PUBACK from client1 (Mid: 2)
知道如何处理这种情况吗？
编辑：
按照接受的答案中提出的解决方法实施时，出现以下错误。我的 Spring 上下文是从网络应用加载的。我试过将 IgnoreUnsubscribePahoClientFactory 放在一个单独的 JAR（与 spring-integration / paho 相同级别）以及 webapp 类 本身中。
2015-12-02 15:47:43,703 ERROR org.springframework.integration.handler.LoggingHandler - org.springframework.aop.framework.AopConfigException: Could not generate CGLIB subclass of class [class org.eclipse.paho.client.mqttv3.MqttAsyncClient]: Common causes of this problem include using a final class or a non-visible class; nested exception is org.springframework.cglib.core.CodeGenerationException: java.lang.reflect.InvocationTargetException-->null
at org.springframework.aop.framework.CglibAopProxy.getProxy(CglibAopProxy.java:206)
at org.springframework.aop.framework.ProxyFactoryBean.getProxy(ProxyFactoryBean.java:368)
at org.springframework.aop.framework.ProxyFactoryBean.getSingletonInstance(ProxyFactoryBean.java:322)
at org.springframework.aop.framework.ProxyFactoryBean.getObject(ProxyFactoryBean.java:246)
at com.ecs.vdm.rest.integration.IgnoreUnsubscribePahoClientFactory.proxy(IgnoreUnsubscribePahoClientFactory.java:62)
at com.ecs.vdm.rest.integration.IgnoreUnsubscribePahoClientFactory.getAsyncClientInstance(IgnoreUnsubscribePahoClientFactory.java:43)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.connectAndSubscribe(MqttPahoMessageDrivenChannelAdapter.java:216)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.access0(MqttPahoMessageDrivenChannelAdapter.java:45)
at org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter.run(MqttPahoMessageDrivenChannelAdapter.java:272)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:304)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:178)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: org.springframework.cglib.core.CodeGenerationException: java.lang.reflect.InvocationTargetException-->null
at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:237)
at org.springframework.cglib.proxy.Enhancer.createHelper(Enhancer.java:377)
at org.springframework.cglib.proxy.Enhancer.createClass(Enhancer.java:317)
at org.springframework.aop.framework.ObjenesisCglibAopProxy.createProxyClassAndInstance(ObjenesisCglibAopProxy.java:57)
at org.springframework.aop.framework.CglibAopProxy.getProxy(CglibAopProxy.java:202)
... 16 more
Caused by: java.lang.reflect.InvocationTargetException
at sun.reflect.GeneratedMethodAccessor37.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:606)
at org.springframework.cglib.core.ReflectUtils.defineClass(ReflectUtils.java:384)
at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:219)
... 20 more
Caused by: java.lang.SecurityException: class "org.eclipse.paho.client.mqttv3.MqttAsyncClient$$EnhancerBySpringCGLIB$$d14754a9_4603"'s signer information does not match signer information of other classes in the same package
at java.lang.ClassLoader.checkCerts(ClassLoader.java:952)
at java.lang.ClassLoader.preDefineClass(ClassLoader.java:666)
at java.lang.ClassLoader.defineClass(ClassLoader.java:794)
... 25 more
这是一个错误 - 它在 stop() 期间无条件取消订阅。
我没有看到简单的解决方法；但我有几个想法；我会post这里if/when我有事
同时，请打开一个JIRA Issue。
编辑
这有点像大锤，但它应该适合你；它有效地忽略了客户端上对 unsubscribe 的调用。它可以做得更复杂一些，只在 QOS > 0 时忽略，但这会涉及更多。
如果您已经在使用 DefaultMqttPahoClientFactory，只需将 bean class 更改为这个。如果您当前未使用工厂，请将其声明为 bean 并使用 client-factory 属性将其提供给适配器。
我们会在即将发布的版本中妥善修复它。