public void publishTopic(String topicStr, String pubMsg, int pubQoS, boolean retained) throws Exception
{
    if (!connect())
    {
        PV.logger.log(Level.WARNING, "Could not publish to mqtt topic \"" + topicStr
                + "\" due to no broker connection");
        throw new Exception("MQTT publish failed: no broker connection");
    }

    MqttTopic topic = myClient.getTopic(topicStr);
    MqttMessage message = new MqttMessage(pubMsg.getBytes());
    message.setQos(pubQoS);
    message.setRetained(retained);

    MqttDeliveryToken token = null;
    try {
        // publish message to broker
        token = topic.publish(message);
        // Wait until the message has been delivered to the broker
        token.waitForCompletion();
        Thread.sleep(100);
    } catch (Exception ex) {
        throw new Exception("Failed to publish message to broker", ex);
    }
}
 

/**
 * 发送一个心跳包，保持长连接
 * @return MqttDeliveryToken specified token you can choose to wait for completion
 */
private synchronized MqttDeliveryToken sendKeepAlive()
        throws MqttConnectivityException, MqttPersistenceException, MqttException {
    if(!isConnected())
        throw new MqttConnectivityException();

    if(mKeepAliveTopic == null) {
        mKeepAliveTopic = mClient.getTopic(TOPIC);

    }
    Log.i(DEBUG_TAG,"Sending Keepalive to " + MQTT_BROKER);

    MqttMessage message = new MqttMessage(MQTT_KEEP_ALIVE_MESSAGE.getBytes());
    message.setQos(MQTT_KEEP_ALIVE_QOS);
    /**发送一个心跳包给服务器，然后回调到：messageArrived 方法中*/
   return mKeepAliveTopic.publish(message);
}
 

/**
 * Sends a Keep Alive message to the specified topic
 * @see MQTT_KEEP_ALIVE_MESSAGE
 * @see MQTT_KEEP_ALIVE_TOPIC_FORMAT
 * @return MqttDeliveryToken specified token you can choose to wait for completion
 */
private synchronized MqttDeliveryToken sendKeepAlive()
throws MqttConnectivityException, MqttPersistenceException, MqttException {
        if(!isConnected())
                throw new MqttConnectivityException();

        if(mKeepAliveTopic == null) {
                mKeepAliveTopic = mClient.getTopic(
                        String.format(Locale.US, MQTT_KEEP_ALIVE_TOPIC_FORAMT,mDeviceId));
        }

        Log.i(DEBUG_TAG,"Sending Keepalive to " + MQTT_BROKER);

        MqttMessage message = new MqttMessage(MQTT_KEEP_ALIVE_MESSAGE);
        message.setQos(MQTT_KEEP_ALIVE_QOS);

        return mKeepAliveTopic.publish(message);
}
 

/**
 * Start a registered producer, so that it can start sending messages.
 * 
 * @param publisher
 *            to start
 */
private void startProducer(MqttMessageProducer publisher) {

    logger.trace("Starting message producer for broker '{}'", name);

    publisher.setSenderChannel(new MqttSenderChannel() {

        @Override
        public void publish(String topic, byte[] payload) throws Exception {

            if (!started) {
                logger.warn("Broker connection not started. Cannot publish message to topic '{}'", topic);
                return;
            }

            // Create and configure a message
            MqttMessage message = new MqttMessage(payload);
            message.setQos(qos);
            message.setRetained(retain);

            // publish message asynchronously
            MqttTopic mqttTopic = client.getTopic(topic);
            MqttDeliveryToken deliveryToken = mqttTopic.publish(message);

            logger.debug("Publishing message {} to topic '{}'", deliveryToken.getMessageId(), topic);
            if (!async) {
                // wait for publish confirmation
                deliveryToken.waitForCompletion(10000);
                if (!deliveryToken.isComplete()) {
                    logger.warn(
                            "Did not receive completion message within timeout limit while publishing to topic '{}'",
                            topic);
                }
            }

        }
    });

}
 

/**
 * Publish Message Completion
 */
@Override
public void deliveryComplete(MqttDeliveryToken arg0) {

}