Java阿里云MQTT服务器推送代码
<p>推送消息内容byteMessage的编码方式，可以参考[Java云喇叭云打印协议编码](<a href="http://doc.szzkc.com/web/#/18?page_id=74">http://doc.szzkc.com/web/#/18?page_id=74</a> "Java云喇叭云打印协议编码")</p>
<pre><code class="language-java">package com.aliyun.openservices.lmq.example.demo;
import com.aliyun.openservices.lmq.example.util.ConnectionOptionWrapper;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.xml.ws.handler.MessageContext;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
/**
* 本代码提供签名鉴权模式下 MQ4IOT 客户端发送消息到 MQ4IOT 客户端的示例，其中初始化参数请根据实际情况修改
* 签名模式即使用阿里云账号系统提供的 AccessKey 和 SecretKey 对每个客户端计算出一个独立的签名供客户端识别使用。
* 对于实际业务场景使用过程中，考虑到私钥 SecretKey 的隐私性，可以将签名过程放在受信任的环境完成。
*
* 完整 demo 工程，参考https://github.com/AliwareMQ/lmq-demo
*/
public class MQTTSendP2PMessage {
public static void main(String[] args) throws Exception {
/**
* MQ4IOT 实例 ID，购买后控制台获取
*/
String instanceId = &quot;XXXXX&quot;;
/**
* 接入点地址，购买 MQ4IOT 实例，且配置完成后即可获取，接入点地址必须填写分配的域名，不得使用 IP 地址直接连接，否则可能会导致客户端异常。
*/
String endPoint = &quot;XXXXX.mqtt.aliyuncs.com&quot;;
/**
* 账号 accesskey，从账号系统控制台获取
*/
String accessKey = &quot;XXXXX&quot;;
/**
* 账号 secretKey，从账号系统控制台获取，仅在Signature鉴权模式下需要设置
*/
String secretKey = &quot;XXXXX&quot;;
/**
* MQ4IOT clientId，由业务系统分配，需要保证每个 tcp 连接都不一样，保证全局唯一，如果不同的客户端对象（tcp 连接）使用了相同的 clientId 会导致连接异常断开。
* clientId 由两部分组成，格式为 GroupID@@@DeviceId，其中 groupId 在 MQ4IOT 控制台申请，DeviceId 由业务方自己设置，clientId 总长度不得超过64个字符。
*/
String clientId = &quot;GID_XXXXX@@@IMEI&quot;;
//发送端ClientID，请勿与TOPIC中ClientID重复，
String masterClientId=&quot;GID_XXXXX@@@20191010master&quot;;
/**
* MQ4IOT 消息的一级 topic，需要在控制台申请才能使用。
* 如果使用了没有申请或者没有被授权的 topic 会导致鉴权失败，服务端会断开客户端连接。
*/
final String parentTopic = &quot;XXXXX&quot;;
/**
* QoS参数代表传输质量，根据设备对接协议设定
https://help.aliyun.com/document_detail/42420.html?spm=a2c4g.11186623.6.544.1ea529cfAO5zV3
*/
final int qosLevel = 2;
ConnectionOptionWrapper connectionOptionWrapper = new ConnectionOptionWrapper(instanceId, accessKey, secretKey, masterClientId);
final MemoryPersistence memoryPersistence = new MemoryPersistence();
/**
* 客户端使用的协议和端口必须匹配，具体参考文档 https://help.aliyun.com/document_detail/44866.html?spm=a2c4g.11186623.6.552.25302386RcuYFB
* 如果是 SSL 加密则设置ssl://endpoint:8883
*/
final MqttClient mqttClient = new MqttClient(&quot;tcp://&quot; + endPoint + &quot;:1883&quot;, masterClientId, memoryPersistence);
/**
* 客户端设置好发送超时时间，防止无限阻塞
*/
mqttClient.setTimeToWait(5000);
mqttClient.setCallback(new MqttCallbackExtended() {
@Override
public void connectComplete(boolean reconnect, String serverURI) {
/**
* 客户端连接成功后就需要尽快订阅需要的 topic
*/
System.out.println(&quot;connect success&quot;);
}
@Override
public void connectionLost(Throwable throwable) {
throwable.printStackTrace();
}
@Override
public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
/**
* 消费消息的回调接口，需要确保该接口不抛异常，该接口运行返回即代表消息消费成功。
* 消费消息需要保证在规定时间内完成，如果消费耗时超过服务端约定的超时时间，对于可靠传输的模式，服务端可能会重试推送，业务需要做好幂等去重处理。超时时间约定参考限制
* https://help.aliyun.com/document_detail/63620.html?spm=a2c4g.11186623.6.546.229f1f6ago55Fj
*/
System.out.println(
&quot;receive msg from topic &quot; + s + &quot; , body is &quot; + new String(mqttMessage.getPayload()));
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
System.out.println(&quot;send msg succeed topic is : &quot; + iMqttDeliveryToken.getTopics()[0]);
}
});
mqttClient.connect(connectionOptionWrapper.getMqttConnectOptions());
/**
* MQ4IoT支持点对点消息，即如果发送方明确知道该消息只需要给特定的一个设备接收，且知道对端的 clientId，则可以直接发送点对点消息。
* 点对点消息不需要经过订阅关系匹配，可以简化订阅方的逻辑。点对点消息的 topic 格式规范是 {{parentTopic}}/p2p/{{targetClientId}}
*/
final String p2pSendTopic = parentTopic + &quot;/p2p/&quot; + clientId;
byte[] byteMessage=new byte[]{0x1E,0x10,0x03,0x00,0x0a,0x41,0x42,0x43,0x44,0x0d,0x0a,0x1B,0x63,0x6f,(byte)0x90};
MqttMessage message = new MqttMessage(byteMessage);
message.setQos(qosLevel);
mqttClient.publish(p2pSendTopic, message);
Thread.sleep(Long.MAX_VALUE);
}
}
</code></pre>