一、问题描述：

1、弱网环境
2、使用MqttTopic.pulish方法，并返回一个token，线程无限阻塞在publish导致后续方法无法执行
具体代码如下：

 MqttTopic mqttTopic = client.getTopic(topic);
    if (null == mqttTopic) {
      log.error("topic not exist");
      return;
    }
 try {
     log.warn("13(1)-------------mqttTopic.publish消息开始");
     MqttDeliveryToken token  = mqttTopic.publish(message);
     token.waitForCompletion(100);   //设置超时时间
     log.info("mqttTopic.publish:{}{}{}{}{}{}{}...",pushMessage[0],pushMessage[1],pushMessage[2],pushMessage[3],pushMessage[4],pushMessage[5],pushMessage[6]);
     log.warn("13(2)-------------mqttTopic.publish消息结束");
 } catch (Exception e) {
      log.error("mqttTopic.publish:推送消息超时100ms");
      throw new RuntimeException(e);
    }    

二、问题分析：

publish具体的代码

具体深入这个publish方法，可以看到具体代码是

public MqttDeliveryToken publish(MqttMessage message) throws MqttException, MqttPersistenceException {
        MqttDeliveryToken token = new MqttDeliveryToken(this.comms.getClient().getClientId());
        token.setMessage(message);
        this.comms.sendNoWait(this.createPublish(message), token);
        token.internalTok.waitUntilSent();
        return token;
    }

重点在于token.internalTok.waitUntilSent();，再次深入查看代码发现

token.internalTok.waitUntilSent()的具体代码

 synchronized(this.sentLock) {
            synchronized(this.responseLock) {
                if (this.exception != null) {
                    throw this.exception;
                }
            }
            while(!this.sent) {
                try {
                    this.log.fine(CLASS_NAME, "waitUntilSent", "409", new Object[]{this.getKey()});
                    this.sentLock.wait();
                } catch (InterruptedException var4) {
                }
            }
            if (!this.sent) {
                if (this.exception == null) {
                    throw ExceptionHelper.createMqttException(6);
                } else {
                    throw this.exception;
                }
            }

代码使用了多个状态标志位去判断消息是否发送、接受到结果，否则就一直无限循环，这个就是问题的关键所在。

可能造成无限阻塞的情况：

·①、弱网环境。发送期间发送了网络断开的情况，期间不断等待收到回复的消息的解锁，重连后服务端和客户端状态不一致，导致一直无法解锁，造成阻塞；
②、多线程竞争。考虑下项目中是否有多个地方使用client进行同时publish消息，导致没有notifyAll()；思路来源自这个博文。具体原话是：

所以猜测是没有线程唤醒本线程，那么什么时候才会有线程唤醒本线程？
猜测： this.comms.sendNoWait(this.createPublish(message), token); 中有taskExecutor去执行“将消息发送到队列中”task，task执行完成后，会将sent=true，然后notifyAll()。然而，在多线程的时候，this.comms.sendNoWait(this.createPublish(message), token)的this是同一个MqttTopic时（topic名相同时），线程1等待时被线程2抢到cpu执行权，线程2会将taskExecutor执行的task更新为线程2的task，线程1再执行时没有了task，不会notifyAll()，就会一直等待下去了。

修改方式：

①、暂时替换成简易的

 private MqttClient client = new MqttClient();
 client.setTimeToWait(100);	//设置超时100ms
 
 client.publish(topic,message);

原因是 client.publish代码中没有使用了token.internalTok.waitUntilSent()进行阻塞，而是简单暴力的使用MqttAsyncClient发送，并设置超时等待，简单有效的避免这个问题。

 this.aClient.publish(topic, message, (Object)null, (IMqttActionListener)null).waitForCompletion(this.getTimeToWait());

②、上锁，防止多线程竞争

 synchronized (this) {
      try {
               token = mqttTopic.publish(message);
               token.waitForCompletion(1000L);
           } catch (MqttPersistenceException e) {
                e.printStackTrace();
           } catch (MqttException e) {
                e.printStackTrace();
           }