1. public void run() {
2. final String methodName = "run";
3. MqttWireMessage message = null;
4. while (running && (out != null)) {//超级无限循环
5. try {
6. message = clientState.get();//主要看这里获取message时进行了阻塞，即clientState.get()方法没有获得消息的时候，代码一直处理阻塞状态，不会一直无限循环！
7. if (message != null) {
8. //@TRACE 802=network send key={0} msg={1}
9. log.fine(className,methodName,"802", new Object[] {message.getKey(),message});
10. if (message instanceof MqttAck) {
11. out.write(message);
12. out.flush();
13. } else {
14. MqttToken token = tokenStore.getToken(message);
15. // While quiescing the tokenstore can be cleared so need
16. // to check for null for the case where clear occurs
17. // while trying to send a message.
18. if (token != null) {
19. synchronized (token) {//使用了同步，防止一次性多个写操作。
20. out.write(message);
21. out.flush();
22. clientState.notifySent(message);//通知已经发送了一个消息
23. }
24. }
25. }
26. } else { // null message
27. //@TRACE 803=get message returned null, stopping}
28. log.fine(className,methodName,"803");
29. running = false;
30. }
31. } catch (MqttException me) {
32. handleRunException(message, me);
33. } catch (Exception ex) {
34. handleRunException(message, ex);
35. }
36. } // end while
37. //@TRACE 805=<
38. log.fine(className, methodName,"805");
39. }

总 的过程如下：send方法将消息放入到pendingMessages队列和pendingFlows当中同时发送消息唤醒等待中的线程，get等待 pendingMessages队列和pendingFlows中的消息，同时等待唤醒，如果有消息放入，同时被唤醒，那么就执行发送消息的操作。这个过 程是不是跟操作系统当中的生产者-消费者的关系一样呢！！！

1. public void run() {
2. final String methodName = "run";
3. MqttToken token = null;
4. //在这里，因为客户端无法判断，服务器什么时候能够发消息过来，因此只能采用无限循环的方式，不断的去判断是否有新消息发送过来。
5. while (running && (in != null)) {//超级无限循环
6. try {
7. //@TRACE 852=network read message
8. log.fine(className,methodName,"852");
9. MqttWireMessage message = in.readMqttWireMessage();// 这里，因为socket.getInputStream()一直在阻塞，如果没有消息是读不到message的，因此在这里的while循环也没有无限制 的运行下去，只有在有消息的时候才往下走。socket默认是阻塞的，就是在读的时候如果读不到资源就会一直等待，直到超时（如果设置了超时时间的话）， 如果服务端和客户端都在读的话而没有写的话就会一直阻塞。你可以使用SocketChannel，设置socket的通道，使其变成非阻塞的。
10. if (message instanceof MqttAck) {//判断是否是确认包
11. token = tokenStore.getToken(message);
12. if (token!=null) {
13. synchronized (token) {
14. // Ensure the notify processing is done under a lock on the token
15. // This ensures that the send processing can complete  before the
16. // receive processing starts! ( request and ack and ack processing
17. // can occur before request processing is complete if not!
18. clientState.notifyReceivedAck((MqttAck)message);
19. }
20. } else {
21. // It its an ack and there is no token then something is not right.
22. // An ack should always have a token assoicated with it.
23. throw new MqttException(MqttException.REASON_CODE_UNEXPECTED_ERROR);
24. }
25. } else {
26. // A new message has arrived，一个新消息过来。
27. clientState.notifyReceivedMsg(message);//点击进入之后
28. }
29. }
30. catch (MqttException ex) {
31. //@TRACE 856=Stopping, MQttException
32. log.fine(className,methodName,"856",null,ex);
33. running = false;
34. // Token maybe null but that is handled in shutdown
35. clientComms.shutdownConnection(token, ex);
36. }
37. catch (IOException ioe) {
38. //@TRACE 853=Stopping due to IOException
39. log.fine(className,methodName,"853");
40. running = false;
41. // An EOFException could be raised if the broker processes the
42. // DISCONNECT and ends the socket before we complete. As such,
43. // only shutdown the connection if we're not already shutting down.
44. if (!clientComms.isDisconnecting()) {
45. clientComms.shutdownConnection(token, new MqttException(MqttException.REASON_CODE_CONNECTION_LOST, ioe));
46. } // else {
47. }
48. }
49. //@TRACE 854=<
50. log.fine(className,methodName,"854");
51. }

1. if (message instanceof MqttPublish) {
2. MqttPublish send = (MqttPublish) message;
3. switch (send.getMessage().getQos()) {
4. case 0:
5. case 1:
6. if (callback != null) {
7. callback.messageArrived(send);
8. }
9. break;

1. public void messageArrived(MqttPublish sendMessage) {
2. final String methodName = "messageArrived";
3. if (mqttCallback != null) {
4. // If we already have enough messages queued up in memory, wait
5. // until some more queue space becomes available. This helps
6. // the client protect itself from getting flooded by messages
7. // from the server.
8. synchronized (spaceAvailable) {
9. if (!quiescing && messageQueue.size() >= INBOUND_QUEUE_SIZE) {
10. try {
11. // @TRACE 709=wait for spaceAvailable
12. log.fine(className, methodName, "709");
13. spaceAvailable.wait();
14. } catch (InterruptedException ex) {
15. }
16. }
17. }
18. if (!quiescing) {
19. messageQueue.addElement(sendMessage);
20. // Notify the CommsCallback thread that there's work to do...
21. synchronized (workAvailable) {
22. // @TRACE 710=new msg avail, notify workAvailable
23. log.fine(className, methodName, "710");
24. workAvailable.notifyAll();
25. }
26. }
27. }
28. }