• android MQTT使用

    MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在 TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议，为此，它需要一个消息中间件 。

    MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

    ，百度文库和官方网站介绍很详细，如下

    MQTT_百度百科 (baidu.com)

    首页 | MQTT中文网 (p2hp.com)

    简单总结一下。使用订阅/发布模式，将消息的发送方和接受方解耦。mqtt分别由发布（发送方）和订阅（接口方）、MQTT Broker（代理服务器）组成。代理服务器安装在我们的服务器。智能硬件设备是发布也是订阅。同样APP端或小程序等客户端也是发布和订阅。硬件上报数据到服务器，服务器发到APP，这个时候硬件是发布，而APP是订阅，如果说APP发一条指令给智能硬件，app是发布，智能硬件是订阅。

    一、选择下载代理服务器及启动

    MQTT代理服务器有很多种，这里使用emqx,部署在windown系统上。也可以部署在linnux。

    1、首先下载emqx,下载链接如下：

    Download EMQX

    选择下载后，保存到电脑磁盘，解压出来，不在放在带中文目录。如图下

    2、进入bin目录，复制目录路径

    3、 启动emqx，cmd打开命令行窗口，输入D:回车，然后进入复制的目录路径进入，emqt start启动，如下

    cd D:\emqx-5.0.0-otp24.2.1-windows-amd64\bin

    emqt start

    4、启动emqt 后，我们打开管理页面，输入http://192.168.2.6:18083/#/login?to=/subscriptions/subscription

    也可以http://127.0.0.1:18083/#/login?to=/subscriptions/subscription 

    cmd打开命令窗口，输入ipconfig回车，查看本机电脑IP，通过局域网ip方式访问。我们使用APP就得通过局域网IP进行通信，如下会介绍，在浏览器打开链接后，可以看到一个登录页面，输入账号密码登录可以进入首页，账号默认是admin，密码默认public，登录后可以看到的页面如下

    点击这里可以查看到已经连接的客户端

     打开这里可以看到主题订阅

    打开这里设置显示的语言

     二、客户端测试工具mqttfx使用

    下载地址如下

    Download (jensd.de)

    1、我们百度下载mqttfx-1.7.1-windows-x64进行安装到电脑或服务器上，我这里测试的，下载windown 版本的，安装完成打开如下。

     2、查看发布（发送方）使用

     3、查看订阅（查看订阅数据） 

    三、手动写安卓APP代码实现发布和订阅（发送数据和接收数据）

    使用android stutio 开发工具，使用java开发

    1、添加权限

    1.     <uses-permission android:name="android.permission.INTERNET" />
    2.     <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    3.     <uses-permission android:name="android.permission.WAKE_LOCK" />

    2、在build.gradle添加包

    1. plugins {
    2.     id 'com.android.application'
    3. }
    4.  
    5. android {
    6.     compileSdk 32
    7.  
    8.     defaultConfig {
    9.         applicationId "com.nyw.mqttdemo"
    10.         minSdk 21
    11.         targetSdk 32
    12.         versionCode 1
    13.         versionName "1.0"
    14.  
    15.         testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"
    16.     }
    17.  
    18.     buildTypes {
    19.         release {
    20.             minifyEnabled false
    21.             proguardFiles getDefaultProguardFile('proguard-android-optimize.txt'), 'proguard-rules.pro'
    22.         }
    23.     }
    24.     compileOptions {
    25.         sourceCompatibility JavaVersion.VERSION_1_8
    26.         targetCompatibility JavaVersion.VERSION_1_8
    27.     }
    28. }
    29.  
    30. dependencies {
    31.  
    32.     implementation 'androidx.appcompat:appcompat:1.3.0'
    33.     implementation 'com.google.android.material:material:1.4.0'
    34.     implementation 'androidx.constraintlayout:constraintlayout:2.0.4'
    35.     testImplementation 'junit:junit:4.13.2'
    36.     androidTestImplementation 'androidx.test.ext:junit:1.1.3'
    37.     androidTestImplementation 'androidx.test.espresso:espresso-core:3.4.0'
    38.     //添加mqtt 2个包
    39.     implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.1.0'
    40.     implementation 'org.eclipse.paho:org.eclipse.paho.android.service:1.1.1'
    41. }

    在settings.gradle添加包下载链接

    maven {
    url "https://repo.eclipse.org/content/repositories/paho-snapshots/"
}
    1. pluginManagement {
    2.     repositories {
    3.         gradlePluginPortal()
    4.         google()
    5.         mavenCentral()
    6.         maven {
    7.             url "https://repo.eclipse.org/content/repositories/paho-snapshots/"
    8.         }
    9.     }
    10. }
    11. dependencyResolutionManagement {
    12.     repositoriesMode.set(RepositoriesMode.FAIL_ON_PROJECT_REPOS)
    13.     repositories {
    14.         google()
    15.         mavenCentral()
    16.     }
    17. }
    18. rootProject.name = "MQTTDemo"
    19. include ':app'

    3、页面

    1. <?xml version="1.0" encoding="utf-8"?>
    2. <androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    3.     xmlns:app="http://schemas.android.com/apk/res-auto"
    4.     xmlns:tools="http://schemas.android.com/tools"
    5.     android:layout_width="match_parent"
    6.     android:layout_height="match_parent"
    7.     tools:context=".MainActivity">
    8.  
    9.     <TextView
    10.         android:id="@+id/tv_msg"
    11.         android:layout_width="wrap_content"
    12.         android:layout_height="wrap_content"
    13.         android:text="Hello World!"
    14.         app:layout_constraintBottom_toBottomOf="parent"
    15.         app:layout_constraintLeft_toLeftOf="parent"
    16.         app:layout_constraintRight_toRightOf="parent"
    17.         app:layout_constraintTop_toTopOf="parent" />
    18.  
    19.     <Button
    20.         android:id="@+id/btn_send_data"
    21.         android:layout_width="match_parent"
    22.         android:layout_height="wrap_content"
    23.         android:text="发送数据"
    24.         tools:layout_editor_absoluteX="-46dp"
    25.         tools:layout_editor_absoluteY="131dp" />
    26.  
    27. </androidx.constraintlayout.widget.ConstraintLayout>

    4、java代码 

    1. package com.nyw.mqttdemo;
    2.  
    3. import androidx.appcompat.app.AppCompatActivity;
    4.  
    5. import android.annotation.SuppressLint;
    6. import android.content.ComponentName;
    7. import android.content.Intent;
    8. import android.content.ServiceConnection;
    9. import android.os.Bundle;
    10. import android.os.Handler;
    11. import android.os.IBinder;
    12. import android.os.Message;
    13. import android.view.View;
    14. import android.widget.Button;
    15. import android.widget.TextView;
    16. import android.widget.Toast;
    17.  
    18. import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
    19. import org.eclipse.paho.client.mqttv3.MqttCallback;
    20. import org.eclipse.paho.client.mqttv3.MqttClient;
    21. import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
    22. import org.eclipse.paho.client.mqttv3.MqttException;
    23. import org.eclipse.paho.client.mqttv3.MqttMessage;
    24. import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
    25.  
    26. import java.io.UnsupportedEncodingException;
    27. import java.util.concurrent.Executors;
    28. import java.util.concurrent.ScheduledExecutorService;
    29. import java.util.concurrent.TimeUnit;
    30.  
    31. public class MainActivity extends AppCompatActivity  {
    32.     private Button btn_send_data;
    33.     private TextView tv_msg;
    34.     //访问的地址，这里填写mqtt代理服务器ip
    35.     private String host = "tcp://192.168.2.6:1883";
    36.     //登录账号
    37.     private String userName = "admin";
    38.     //登录密码
    39.     private String passWord = "public";
    40.     //这个是mqtt_id  唯一id，每个客户端应该生成不同的唯一id，这里测试演示数据，我就写固定的id，打包到不同的手机上安装，要手动修改或修改成动态id，获取随机数+时间方式生成一个随机数作为唯一id
    41.     private String mqtt_id="1";
    42.  
    43.     private int i = 1;
    44.  
    45.     private Handler handler;
    46.  
    47.     private MqttClient client;
    48.  
    49.     private String mqtt_sub_topic = "second3"; //订阅主题(消息接收方)
    50.  
    51.     private String mqtt_pub_topic ="first";//发布主题（消息发送方）
    52.  
    53.     private MqttConnectOptions options;
    54.  
    55.     private ScheduledExecutorService scheduler;
    56.  
    57.  
    58.     @Override
    59.     protected void onCreate(Bundle savedInstanceState) {
    60.         super.onCreate(savedInstanceState);
    61.         setContentView(R.layout.activity_main);
    62.  
    63.         inintView();
    64.  
    65.  
    66.     }
    67.  
    68.     private void inintView() {
    69.  
    70.  
    71.         btn_send_data=findViewById(R.id.btn_send_data);
    72.         tv_msg=findViewById(R.id.tv_msg);
    73.         btn_send_data.setOnClickListener(new View.OnClickListener() {
    74.             @Override
    75.             public void onClick(View v) {
    76.                 //发送数据
    77.                 publishmessageplus(mqtt_pub_topic,"我要发布订阅，发送数据");
    78.             }
    79.         });
    80.         init();
    81.         startReconnect();
    82.         handler = new Handler() {
    83.  
    84.             @SuppressLint("SetTextIl8n")
    85.  
    86.             public void handleMessage(Message msg) {
    87.  
    88.                 super.handleMessage(msg);
    89.  
    90.                 switch (msg.what) {
    91.  
    92.                     case 1: //开机校验更新回传
    93.  
    94.                         break;
    95.  
    96.                     case 2: //反馈回转
    97.  
    98.                         break;
    99.  
    100.                     case 3: //MQTT收到消息回传
    101.                         tv_msg.setText(msg.obj.toString());
    102.  
    103.                         break;
    104.  
    105.                     case 30: //连接失败
    106.  
    107.                         Toast.makeText(MainActivity.this,"连接失败",Toast.LENGTH_SHORT).show();
    108.  
    109.                         break;
    110.  
    111.                     case 31: //连接成功
    112.  
    113.                         Toast.makeText(MainActivity.this,"连接成功",Toast.LENGTH_SHORT).show();
    114.  
    115.                         try {
    116.  
    117.                             client.subscribe(mqtt_sub_topic,2);
    118.  
    119.                         } catch (MqttException e) {
    120.  
    121.                             e.printStackTrace();
    122.  
    123.                         }
    124.  
    125. //                        publishmessageplus(mqtt_pub_topic,"第一个客户端发送的信息");
    126.  
    127.  
    128.  
    129.                         break;
    130.  
    131.                     default:
    132.  
    133.                         break;
    134.  
    135.                 }
    136.  
    137.             }
    138.  
    139.         };
    140.  
    141.     }
    142.  
    143.     /**
    144.      * 初始化MQTT配制
    145.      */
    146.     private void init() {
    147.  
    148.         try {
    149.  
    150.             //host为主机名，test为clientid即连接MQTT的客户端ID，一般以客户端唯一标识符表示，MemoryPersistence设置clientid的保存形式，默认为以内存保存
    151.  
    152.             client = new MqttClient(host, mqtt_id,
    153.  
    154.                     new MemoryPersistence());
    155.  
    156.             //MQTT的连接设置
    157.  
    158.             options = new MqttConnectOptions();
    159.  
    160.             //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
    161.  
    162.             options.setCleanSession(true);
    163.  
    164.             //设置连接的用户名
    165.  
    166.             options.setUserName(userName);
    167.  
    168.             //设置连接的密码
    169.  
    170.             options.setPassword(passWord.toCharArray());
    171.  
    172.             // 设置超时时间 单位为秒
    173.  
    174.             options.setConnectionTimeout(10);
    175.  
    176.             // 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
    177.  
    178.             options.setKeepAliveInterval(20);
    179.  
    180.             //设置回调
    181.  
    182.             client.setCallback(new MqttCallback() {
    183.  
    184.                 @Override
    185.  
    186.                 public void connectionLost(Throwable cause) {
    187.  
    188.                     //连接丢失后，一般在这里面进行重连
    189.  
    190.                     System.out.println("connectionLost----------");
    191.  
    192.                 }
    193.  
    194.                 @Override
    195.  
    196.                 public void deliveryComplete(IMqttDeliveryToken token) {
    197.  
    198.                     //publish后会执行到这里
    199.  
    200.                     System.out.println("deliveryComplete---------"
    201.  
    202.                             + token.isComplete());
    203.  
    204.                 }
    205.  
    206.                 @Override
    207.  
    208.                 public void messageArrived(String topicName, MqttMessage message)
    209.  
    210.                         throws Exception {
    211.  
    212.                     //subscribe后得到的消息会执行到这里面
    213.  
    214.                     System.out.println("messageArrived----------");
    215.  
    216.                     Message msg = new Message();
    217.  
    218.                     msg.what = 3;
    219.                     //重新编码，解决乱码问题
    220.                    String strMsg= new String(message.getPayload(),"GB2312");
    221.                     msg.obj = topicName + "---" + strMsg;
    222.  
    223.                     handler.sendMessage(msg);
    224.  
    225.                 }
    226.  
    227.             });
    228.  
    229.         } catch (Exception e) {
    230.  
    231.             e.printStackTrace();
    232.  
    233.         }
    234.  
    235.     }
    236.  
    237.     /**
    238.      * 连接
    239.      */
    240.     private void Mqtt_connect() {
    241.  
    242.         new Thread(new Runnable() {
    243.  
    244.             @Override
    245.  
    246.             public void run() {
    247.  
    248.                 try {
    249.  
    250.                     if (!(client.isConnected())){
    251.  
    252.                         client.connect(options);
    253.  
    254.                         Message msg = new Message();
    255.  
    256.                         msg.what=31;
    257.  
    258.                         handler.sendMessage(msg);
    259.  
    260.                     }
    261.  
    262.  
    263.  
    264.                 } catch (Exception e) {
    265.  
    266.                     e.printStackTrace();
    267.  
    268.                     Message msg = new Message();
    269.  
    270.                     msg.what = 30;
    271.  
    272.                     handler.sendMessage(msg);
    273.  
    274.                 }
    275.  
    276.             }
    277.  
    278.         }).start();
    279.  
    280.     }
    281.  
    282.     /**
    283.      * 线程池执行任务，心跳包，当断开连接后，执行自动连接代理服务器
    284.      */
    285.     private void startReconnect() {
    286.  
    287.         scheduler = Executors.newSingleThreadScheduledExecutor();
    288.  
    289.         scheduler.scheduleAtFixedRate(new Runnable() {
    290.  
    291.             @Override
    292.  
    293.             public void run() {
    294.                 if (!client.isConnected()) {
    295.                     Mqtt_connect();
    296.  
    297.                 }
    298.  
    299.             }
    300.  
    301.         }, 0 * 1000, 10 * 1000, TimeUnit.MILLISECONDS);
    302.  
    303.     }
    304.  
    305.     /**
    306.      * 发布订阅
    307.      * @param topic
    308.      * @param message2
    309.      */
    310.     private void publishmessageplus(String topic,String message2)
    311.  
    312.     {
    313.  
    314.         if (client == null || !client.isConnected()) {
    315.  
    316.             return;
    317.  
    318.         }
    319.  
    320.         MqttMessage message = new MqttMessage();
    321.         message.setPayload(message2.getBytes());
    322.  
    323.         try {
    324.  
    325.             client.publish(topic,message);
    326.  
    327.         } catch (MqttException e) {
    328.  
    329.  
    330.  
    331.             e.printStackTrace();
    332.  
    333.         }
    334.  
    335.     }
    336.  
    337.  
    338.  
    339. }

    5、运行打包APP测试，这里放截图出来了，源码下载链接如下

    androidmqttdemo代码可直接使用-Android文档类资源-CSDN下载

  • 相关阅读:
    PX4 固件常用 QGroundControl 参数设置
    服务器有几种http强制跳转https设置方法
    IntelliJ IDEA使用——插件推荐
    vue学习笔记——简单入门总结(一)
    FPGA串口回环实验
    【Leetcode周赛306】6148、6149、6150、6151
    SpringCloud-Feign
    C语言数组和指针笔试题(二)(一定要看)
    【echarts】06、echarts+vue2 - 雷达图
    2023计算机四非保研（复试：东北大学，成电，西电，浙软，中海洋，天大）
  • 原文地址：https://blog.csdn.net/u013519290/article/details/125409568