Spring MqttPahoMessageDrivenChannelAdapter 연결이 끊어졌습니다 : 연결이 끊어졌습니다; 다시 시도하는

우리는 Spring Message-Driven-Channel-Adapter를 사용하여 MQTT 주제를 등록합니다. 그러나 우리는 매우 자주 오류 아래에 있습니다. 자바 스크립트 클라이언트 (mqttws31.js)를 사용하여 연결을 테스트했으며 정상적으로 작동합니다. 연결에 문제가 없음을 의미합니다.

오류 : -

org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter connectionLost
SEVERE: Lost connection:Connection lost; retrying...

MQTT 메시지 : -

[payload=6483D03E4C75BA943148F18D73,1.00,1E, headers={mqtt_retained=false, mqtt_qos=0, 
id=5fa41168-34c6-1e3d-a775-e3146842990a, mqtt_topic=TEST/GATEWAY2, mqtt_duplicate=false, timestamp=1499067757559}]

구성 : -

<bean id="clientFactory"
    class="org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory">
    <property name="userName" value="${mqtt.username}" />
    <property name="password" value="${mqtt.password}" />
</bean>

<int-mqtt:message-driven-channel-adapter
    id="mqttInbound" client-id="${mqtt.default.client.id}" url="${mqtt.url}"
    topics="${topics}" client-factory="clientFactory" auto-startup="true"
    channel="output" error-channel="errorChannel" />


<int:channel id="output" />
<int:channel id="errorChannel" />

<int:service-activator input-channel="errorChannel"
    ref="errorMessageLogger" method="logError" />
<bean id="errorMessageLogger" class="com.mqtt.ErrorMessageLogger" />

<int:service-activator input-channel="output"
    method="handleMessage" ref="mqttLogger" />
<bean id="mqttLogger" class="com.mqtt.MqttReciever" />

Pom.hml :

<dependency>
    <groupId>org.eclipse.paho</groupId>
    <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
    <version>1.1.1</version>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-mqtt</artifactId>
    <version>4.2.2.RELEASE</version>
</dependency>

org.eclipse.paho.client.mqttv3-1.1.1-sources.jar를 디버깅하는 동안 : -

CommsReceiver.Java

public void run() {
        final String methodName = "run";
        MqttToken token = null;

        while (running && (in != null)) {
            try {
                //@TRACE 852=network read message
                log.fine(CLASS_NAME,methodName,"852");
                receiving = in.available() > 0;
                MqttWireMessage message = in.readMqttWireMessage();
                receiving = false;

                // instanceof checks if message is null
                if (message instanceof MqttAck) {
                    token = tokenStore.getToken(message);
                    if (token!=null) {
                        synchronized (token) {
                            // Ensure the notify processing is done under a lock on the token
                            // This ensures that the send processing can complete  before the
                            // receive processing starts! ( request and ack and ack processing
                            // can occur before request processing is complete if not!
                            clientState.notifyReceivedAck((MqttAck)message);
                        }
                    } else if(message instanceof MqttPubRec || message instanceof MqttPubComp || message instanceof MqttPubAck) {
                        //This is an ack for a message we no longer have a ticket for.
                        //This probably means we already received this message and it's being send again
                        //because of timeouts, crashes, disconnects, restarts etc.
                        //It should be safe to ignore these unexpected messages.
                        log.fine(CLASS_NAME, methodName, "857");
                    } else {
                        // It its an ack and there is no token then something is not right.
                        // An ack should always have a token assoicated with it.
                        throw new MqttException(MqttException.REASON_CODE_UNEXPECTED_ERROR);
                    }
                } else {
                    if (message != null) {
                        // A new message has arrived
                        clientState.notifyReceivedMsg(message);
                    }
                }
            }
            catch (MqttException ex) {
                //@TRACE 856=Stopping, MQttException
                log.fine(CLASS_NAME,methodName,"856",null,ex);
                running = false;
                // Token maybe null but that is handled in shutdown
                clientComms.shutdownConnection(token, ex);
            }
            catch (IOException ioe) {
                //@TRACE 853=Stopping due to IOException
                log.fine(CLASS_NAME,methodName,"853");

                running = false;
                // An EOFException could be raised if the broker processes the
                // DISCONNECT and ends the socket before we complete. As such,
                // only shutdown the connection if we're not already shutting down.
                if (!clientComms.isDisconnecting()) {
                    clientComms.shutdownConnection(token, new MqttException(MqttException.REASON_CODE_CONNECTION_LOST, ioe));
                }
            }
            finally {
                receiving = false;
            }
        }

        //@TRACE 854=<
        log.fine(CLASS_NAME,methodName,"854");
    }

위의 메소드에서 언젠가 in.readMqttWireMessage ()는 IOException을 던집니다. 그래서 catch 블록을 기반으로 clientComms.shutdownConnection (토큰을 사용하여 다시 연결 ...

해결법

1. ==============================

   1.도움이되는 경우에 공유하고 싶었습니다 ... 동일한 예외가 있었고 여기에 언급 된대로 고유 한 클라이언트 ID가 생성되었는지 (MqttAsyncClient.generateClientId ()) 확인하여이를 수정했습니다. https://github.com/eclipse/paho.mqtt.java/issues/207#issuecomment-338246879

   도움이되는 경우에 공유하고 싶었습니다 ... 동일한 예외가 있었고 여기에 언급 된대로 고유 한 클라이언트 ID가 생성되었는지 (MqttAsyncClient.generateClientId ()) 확인하여이를 수정했습니다. https://github.com/eclipse/paho.mqtt.java/issues/207#issuecomment-338246879

2. ==============================

   2.그러나 당신은 아직도 문제를 설명하고 있지 않습니다. 위의 메시지를 표시하므로 제대로 작동해야합니다. Paho가 연결 문제를 감지하고 있습니다. 그것은 재접속 할 Spring 통합을 알린다.

   그러나 당신은 아직도 문제를 설명하고 있지 않습니다. 위의 메시지를 표시하므로 제대로 작동해야합니다. Paho가 연결 문제를 감지하고 있습니다. 그것은 재접속 할 Spring 통합을 알린다.

   응용 프로그램에 ApplicationListener를 추가하여 예외에 대한 완전한 정보를 얻을 수 있습니다.

   @Bean
   public ApplicationListener<?> eventListener() {
       return new ApplicationListener<MqttConnectionFailedEvent>() {
   
           @Override
           public void onApplicationEvent(MqttConnectionFailedEvent event) {
               event.getCause().printStackTrace();
           }
   
       };
   }
   

   결과:

   Connection lost (32109) - java.io.EOFException
       at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:164)
       at java.lang.Thread.run(Thread.java:748)
   Caused by: java.io.EOFException
       at java.io.DataInputStream.readByte(DataInputStream.java:267)
       at org.eclipse.paho.client.mqttv3.internal.wire.MqttInputStream.readMqttWireMessage(MqttInputStream.java:92)
       at org.eclipse.paho.client.mqttv3.internal.CommsReceiver.run(CommsReceiver.java:116)
       ... 1 more
   

   (내가 브로커를 폐쇄했을 때).

   파 호 클라이언트에 문제가 있다고 생각되면 해당 프로젝트에 대해 문제를 제기해야합니다.

from https://stackoverflow.com/questions/44885702/spring-mqttpahomessagedrivenchanneladapter-lost-connectionconnection-lost-retr by cc-by-sa and MIT license