因公司项目需要，最近接入了阿里 MQTT 协议，今天简单写一下如何简单使用：

一、首先是认证，这里的认证方式有动态注册、ID2认证方式和三元组认证，我们使用的是三元组认证；

1、首先根据阿里云官方文档进行配置，这里不详细介绍了：

https://help.aliyun.com/document_detail/96607.html?spm=a2c4g.11186623.6.615.a6922b0aSxaZOK

2、获取四个参数，分别是 productKey、deviceName、deviceSecret、productSecret  ;

// SDK初始化InitManager.init(this, productKey, deviceName, deviceSecret, productSecret, new IDemoCallback() {    @Override    public void onError(AError aError) {        // 初始化失败，初始化失败之后需要用户负责重新初始化        // 如一开始网络不通导致初始化失败，后续网络回复之后需要重新初始化        Log.e("","初始化失败");    }    @Override    public void onInitDone(Object data) {        Log.e("","初始化成功");       // isInitDone = true;    }});

public interface IDemoCallback extends ILinkKitConnectListener{}

二、认证成功后，通过 RRPC 进行订阅

String testTopic = "/ext/rrpc/" + productKey + "/" + deviceName + "/user/get";final MqttRrpcRegisterRequest registerRequest = new MqttRrpcRegisterRequest();registerRequest.topic = testTopic;LinkKit.getInstance().subscribeRRPC(registerRequest, new IConnectRrpcListener() {    @Override    public void onSubscribeSuccess(ARequest aRequest) {        Log.e("","订阅成功");        // 订阅成功        request = aRequest;    }    @Override    public void onSubscribeFailed(ARequest aRequest, AError aError) {        // 订阅失败        Log.e("","订阅失败");    }    @Override    public void onReceived(ARequest aRequest, IConnectRrpcHandle iConnectRrpcHandle) {        // 收到云端下行        Log.e("","收到下行数据");        // 这里根据个人情况来设置        // 如果不一定是json格式，可以参考如下方式回复        MqttPublishRequest rrpcResponse = new MqttPublishRequest();        if (aRequest instanceof MqttRrpcRequest) {            rrpcResponse.topic = ((MqttRrpcRequest) aRequest).topic;        }        rrpcResponse.payloadObj ="{"id":"123", "code":"200"" + ","data":{} }";        LinkKit.getInstance().publish(rrpcResponse, new IConnectSendListener() {            @Override            public void onResponse(ARequest aRequest, AResponse aResponse) {                ALog.d("", "onResponse() called with: aRequest = [" + aRequest + "], aResponse = [" + aResponse + "]");            }            @Override            public void onFailure(ARequest aRequest, AError aError) {                ALog.d("", "onFailure() called with: aRequest = [" + aRequest + "], aError = [" + aError + "]");            }        });    }    @Override    public void onResponseSuccess(ARequest aRequest) {        Log.e("","响应成功");        // RRPC 响应成功    }    @Override    public void onResponseFailed(ARequest aRequest, AError aError) {        Log.e("","响应失败");        // RRPC 响应失败    }});

三、创建消息接收 Notify，并且在onResume()注册，在onPause() 取消；

private IConnectNotifyListener notifyListener = new IConnectNotifyListener() {    @Override    public void onNotify(String s, String s1, AMessage aMessage) {        Log.e("收到下行消息 topic=" ,s1);        String downstreamData = new String((byte[]) aMessage.data);        Log.e("downstreamData",downstreamData);    }    @Override    public boolean shouldHandle(String s, String s1) {        Log.e("",s1);        return true;    }    @Override    public void onConnectStateChange(String s, ConnectState connectState) {        Log.e("",s);    }};

@Overrideprotected void onResume() {    super.onResume();    LinkKit.getInstance().registerOnPushListener(notifyListener);}

@Overrideprotected void onPause() {    super.onPause();      LinkKit.getInstance().unRegisterOnPushListener(notifyListener);    }