MQTT Android 开发（二）编写Android客户端

客户端"/>

MQTT Android 开发（二）编写Android客户端

mqtt服务器搭建请看MQTT Android 开发（一）MQTT 介绍及服务器搭建
先上效果图

上图在测试时，我们需要进入webSocket 页面，直接点击连接即可，不用修改参数。演示的功能是向客户端订阅的主题发送消息，以及订阅客户端发送消息的主题，接收客户端的消息。

开发过程

1.添加依赖

    implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.4'

此时要注意，如果你使用的开发包是Android X ,有可能会出现这个错误

还需要添加下面这个依赖，解决上述问题，为什么会出现这个问题呢？(～￣(OO)￣)ブ我也忘了。反正出现上面的问题加上就OK了，没问题更好。

    implementation 'androidx.localbroadcastmanager:localbroadcastmanager:1.0.0'

我使用了viewBinding，还需要在build.gradle(app)文件下的android 节点下添加下面的代码开启它

  viewBinding {enabled = true}

2.在清单文件中添加必要的权限(过一段时间再看这里面有些权限不知道是不是非必须的了,大家自己测试吧)

    <uses-permission android:name="android.permission.WAKE_LOCK" /><uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" /><uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" /><uses-permission android:name="android.permission.READ_PHONE_STATE" /><uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" /><uses-permission android:name="android.permission.INTERNET" />

3.在清单文件里开启服务

  <service android:name="org.eclipse.paho.android.service.MqttService" />

4.具体代码如下

public class MainActivity extends AppCompatActivity implements View.OnClickListener {private static final String TAG = "MainActivity";//     MQTT 客户端private MqttAndroidClient client;//   消息代理服务器地址  因为我是本地搭建的 所以只能用模拟器进行测试 10.0.2.2 是Android 规定的访问本地电脑的地址 1883 是MQTT非加密协议端口private String serverURL = "tcp://10.0.2.2:1883";//     客户端 ID，用以识别客户端private String clientId = "HelloWorld";//     使用ViewBinding 代替 findViewByid 可忽略private ActivityMainBinding mBinding;//     连接参数设置private MqttConnectOptions mConnectOptions;//     要订阅的主题private String mTopic = "forTest";//     要发送消息的主题private String mTopic_send = "forSend";/** TextView tvLog 是为了展示效果 代替了log* */@Overrideprotected void onCreate(Bundle savedInstanceState) {super.onCreate(savedInstanceState);mBinding = ActivityMainBinding.inflate(getLayoutInflater());setContentView(mBinding.getRoot());initListener();initConnectOptions();}private void initConnectOptions() {
//        连接设置mConnectOptions = new MqttConnectOptions();
//        设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接mConnectOptions.setCleanSession(true);
//        设置连接的用户名mConnectOptions.setUserName("Stone");
//        设置连接的密码mConnectOptions.setPassword("12345".toCharArray());
//        设置连接超时时间 单位为秒mConnectOptions.setConnectionTimeout(3);
//        设置会话心跳时间 单位为秒 服务器会每隔1.5*60秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制mConnectOptions.setKeepAliveInterval(60);
//        设置自动重连mConnectOptions.setAutomaticReconnect(true);
//       setWill方法，如果项目中需要知道客户端是否掉线可以调用该方法。设置最终端口的通知消息mConnectOptions.setWill(mTopic, "close".getBytes(), 2, true);}private void initListener() {mBinding.btnConnect.setOnClickListener(this);mBinding.btnSubscribe.setOnClickListener(this);mBinding.btnDisConnect.setOnClickListener(this);mBinding.btnPublish.setOnClickListener(this);mBinding.btnUnSubscribe.setOnClickListener(this);}@Overridepublic void onClick(View v) {try {switch (v.getId()) {case R.id.btn_connect:
//                  创建客户端client = new MqttAndroidClient(this, serverURL, clientId);
//                  设置回调client.setCallback(new MqttCallback());
//                  开始连接client.connect(mConnectOptions, null, new MqttListener());mBinding.tvLog.setText("开始连接\n");break;case R.id.btn_subscribe:
//                    订阅主题subscribeTopic();break;case R.id.btn_dis_connect:
//                    断开连接if (client != null) {client.disconnect();mBinding.tvLog.append("客户端已断开\n");}break;case R.id.btn_publish:
//                    发布消息publishMessage();break;case R.id.btn_un_subscribe:
//                    取消订阅主题unSubscribe();break;default:break;}} catch (MqttException e) {}}/*** 连接服务器监听*/class MqttListener implements IMqttActionListener {@Overridepublic void onSuccess(IMqttToken asyncActionToken) {mBinding.tvLog.append("onSuccess: 连接成功\n");// TODO: 2020/10/8 连接成功后该干啥干啥}@Overridepublic void onFailure(IMqttToken asyncActionToken, Throwable exception) {mBinding.tvLog.append("onFailure: 连接失败\n");}}/*** 订阅主题** @throws MqttException*/private void subscribeTopic() throws MqttException {client.subscribe(mTopic, 2, null, new IMqttActionListener() {@Overridepublic void onSuccess(IMqttToken asyncActionToken) {mBinding.tvLog.append("onSuccess: 主题" + mTopic + "订阅成功\n");}@Overridepublic void onFailure(IMqttToken asyncActionToken, Throwable exception) {mBinding.tvLog.append("onFailure: 主题" + mTopic + "订阅失败\n");}});}/*** 取消订阅主题*/private void unSubscribe() throws MqttException {if (client != null) {client.unsubscribe(mTopic, null, new IMqttActionListener() {@Overridepublic void onSuccess(IMqttToken asyncActionToken) {mBinding.tvLog.append("取消订阅主题成功\n");}@Overridepublic void onFailure(IMqttToken asyncActionToken, Throwable exception) {mBinding.tvLog.append("取消订阅主题失败\n");}});}}/*** 发布消息** @throws MqttException*/private void publishMessage() throws MqttException {final String msg = "This is a  message";MqttMessage message = new MqttMessage();message.setQos(1);message.setPayload(msg.getBytes());client.publish(mTopic_send, message, null, new IMqttActionListener() {@Overridepublic void onSuccess(IMqttToken asyncActionToken) {mBinding.tvLog.append("onSuccess: 消息" + msg + "成功发送\n");}@Overridepublic void onFailure(IMqttToken asyncActionToken, Throwable exception) {Log.d(TAG, "onFailure: 消息" + msg + "发送失败\n");}});}/*** 订阅主题的回调*/class MqttCallback implements MqttCallbackExtended {@Overridepublic void connectionLost(Throwable cause) {mBinding.tvLog.append("connectionLost: 连接断开，可以做重连操作\n");}@Overridepublic void messageArrived(String topic, MqttMessage message) {mBinding.tvLog.append("messageArrived: 接收到消息==>" + message + "\n");}@Overridepublic void deliveryComplete(IMqttDeliveryToken token) {mBinding.tvLog.append("deliveryComplete: 消息成功到达broker\n");}@Overridepublic void connectComplete(boolean reconnect, String serverURI) {mBinding.tvLog.append("connectComplete: 连接完成\n");}}}

大多情况下，MQTT的逻辑应该放到service里面，这里进行了简化。