（三）AT指令开发mqtt 阿里云的mqtt搭建过程-Arduino中文社区 - Powered by Discuz!
设为首页收藏本站

Arduino中文社区

 找回密码
 立即注册

QQ登录

只需一步，快速开始

Arduino中文社区»首页 › 热门平台与项目 › esp8266/esp32 Arduino › （三）AT指令开发mqtt 阿里云的mqtt搭建过程 ...
返回列表 发新帖
查看: 12090|回复: 6

（三）AT指令开发mqtt 阿里云的mqtt搭建过程

[复制链接]
jiangzhiwei
发表于 2018-12-6 14:34 | 显示全部楼层 |阅读模式
本帖最后由 jiangzhiwei 于 2019-10-16 23:06 编辑

前两篇介绍了如何去连接MQTT，以及MQTT与TCP协议的关系
本篇讲解一下如何在阿里云上搭建mqtt
（注: 近期学业繁忙
本帖暂时停止维护，如有问题恕不能回复，望谅解）
首先进入阿里云官方
https://www.aliyun.com/?utm_content=se_1000301881

点击云服务器ECS  进入购买界面   怎么购买就不说了吧


如果是学生的话可进入     阿里云翼计划   （本人就是学生购买的优惠9.5一个月）
https://promotion.aliyun.com/ntms/act/campus2018.html
    实名学生验证过后可以优惠很多
 Y79AZ{1LI9Q7DP[(][KOIIA.png

   购买之后右上角进入控制台管理你的云服务器

         进入点击   右侧的   “实例”管理 云服务器  ，
 06LO5S@CXMHY[AJ[A6S1~(J.png U$]JUQ%AXH(HD{K%I6I}UH3.png PBW%9TY6UG5ZPX2AAVTDKW0.png
得到你的公网ip， 6E9RM@5{$K]JY`WRO[RY{0V.png ，然后远程桌面连接
 79@X37F1[9(%QZQOSDZQATH.png
/****下面来介绍搭建MQTT服务器**********************/


http://activemq.apache.org/apollo/download.html   进入下载


根据自己购买的服务的系统选择版本类型。我的是windows  ，所以选择第二个，下载下载好后解压
S]({}SM7_KJNMZBC5CJB858.png


注意          在配置apollo 之前，还需配置java 环境变量   前往
https://www.oracle.com/technetwork/java/javase/downloads/index.html   

自行下载sdk   并配置环境变量   
https://www.cnblogs.com/smyhvae/p/3788534.html   环境变量配置




配置完java环境变量之后，将下载的apollo  解压到电脑上，打开win+R   运行cmd

  先进入所在目录的bin文件夹 (`}{7)H9BQ473B_[%}IO062.png
之后再输入 apollo create mybroker                 即创建一个名为mybroker 的文件夹

创建成功会出现以下提示并多了一个mybroker的文件夹
22ALIX6UHF%({ABE00Z{TWA.png D{336264G_J8XBVJP_FQOAX.png


之后  用命令行进入该文件夹  下的  bin 目录   ，  
~UBC3Z}EE%JSEFF}O`9%M_Q.png    最后运行apollo-broker.cmd 出现如下 7HECQ4]7I7)}_H5IV%@}9Y7.png G`OL5(5P]Q5]EVNF1SJKB}W.png     请留意箭头处所指的IP和端口号


这样就配置成功了   打开

后台Web管理页面：https://127.0.0.1:61681/或http://127.0.0.1:61680/

登录服务器


     



最后在阿里云的安全组里添加  入站规则   添加自己刚刚cmd命令行显示的所指的TCP的 端口号  一般的话是61613
@)T64@{Z9%$P$TXWJ]STN03.png



最后连接测试成功
H978%]D6UF[EC5GC$RRSNU0.png





/********************************下面我来贴上我的AT指令开发MQTT的代码*******************************************************/

不过一切前提是你能建立TCP连接

主要分为2大块
一，建立TCP客户端连接

二，MQTT连接包封装函数   这是我从找到的32 源码的上面移植的封装函数，连接包的格式上有一个地方小错误，这边已经改正了，测试能正常连接



QQ图片20181206142821.png QQ图片20181206142808.png     这边传感器没有接线上去，但是数据传输成功了！
















[mw_shl_code=arduino,true] /*
  * ESP8266 接线
  * vcc接5v
  * RX TX和arduino反接
  *
*/
#include <SoftwareSerial.h>

#define DEBUG
#define _baudrate   9600
// USB TTL 检错用
const int TrigPin =3;
const int EchoPin =4;

#define _rxpin      5
#define _txpin      6
#define    u8   unsigned char
#define    u16   unsigned short
#define    u32  unsigned int

#define    pub  "mydata"//发布的主题
#define    des  "my"  //定义的主题

void start_timer(int t);
void stop_timer();
SoftwareSerial debug( _rxpin, _txpin ); // RX, TX

#define dht_dpin A0// DHT温湿度传感器

#define qiti A3//烟雾传感器接口

boolean isconnect=false;
u8 count=0;
byte bGlobalErr;
byte dht_dat[5];
int MODE=1;
boolean is_send=true;
u8 mqtt_msg[200]={0};

// ESP8266 Wifi模块
#define SSID "esp"//热点名称
#define PASS "12345678"//热点密码
#define IP "172.20.10.5"  //内网客户端模式
#define IP_mqtt "0.0.0.0"//外网mqtt客户端IP

int val=0;//定义数字变量
int cal=0;
// 使用 GET格式
String GET = "\rRead sensor: OK\n";
void  keepAlive();
void(* resetFunc) (void) = 0; //制造重启命令
void setup(){
   InitDHT();
    Serial.begin( _baudrate );
    debug.begin( _baudrate );
     while (!Serial){
  }

    debug.println("AT+RST");
    delay(500);
    while(debug.read()>= 0){}
    debug.listen();
    sendDebug("AT");
    delay(1000);
    if(debug.find("OK"))
    {
        Serial.println("RECEIVED: OK\nData ready to sent!");
         debug.listen();
    }
    delay(100);
    while(Serial.read()>= 0){}
    while(debug.read()>= 0){}
    debug.listen();
}
void loop() {
SELECT_MODE();
//stop_timer();
}


void sendDebug(String cmd)
{
    debug.println(cmd);
    Serial.println(cmd);
}

void InitDHT(){
   // 初始化DHT
   pinMode(dht_dpin,OUTPUT);
   digitalWrite(dht_dpin,HIGH);
}
void ReadDHT(){
is_send=false;
bGlobalErr=0;
byte dht_in;
byte i;
digitalWrite(dht_dpin,LOW);
delay(20);

digitalWrite(dht_dpin,HIGH);
delayMicroseconds(40);
pinMode(dht_dpin,INPUT);
dht_in=digitalRead(dht_dpin);
if(dht_in){
   bGlobalErr=1;
   return;
}
delayMicroseconds(80);
dht_in=digitalRead(dht_dpin);
if(!dht_in){
   bGlobalErr=2;
   return;
   }
delayMicroseconds(80);
for (i=0; i<5; i++)
   dht_dat = read_dht_dat();
pinMode(dht_dpin,OUTPUT);
digitalWrite(dht_dpin,HIGH);
byte dht_check_sum =
       dht_dat[0]+dht_dat[1]+dht_dat[2]+dht_dat[3];
if(dht_dat[4]!= dht_check_sum)
   {bGlobalErr=3;}
};
byte read_dht_dat(){
  byte i = 0;
  byte result=0;
  for(i=0; i< 8; i++){
      while(digitalRead(dht_dpin)==LOW);
      delayMicroseconds(30);
      if (digitalRead(dht_dpin)==HIGH)
     result |=(1<<(7-i));
    while (digitalRead(dht_dpin)==HIGH);
    }
   return result;
}
/***************************************MQTT客户端*************************************/

/*************MQTT连接包*******************/  
u8 mqtt_connect_message(u8 *mqtt_message,char *client_id,char *username,char *password)
{
  Serial.println("  is to connect mqtt");
  u8 client_id_length = strlen(client_id);
  u8 username_length = strlen(username);
  u8 password_length = strlen(password);
  u8 packetLen;
  u8 i,baseIndex;

  packetLen = 12 + 2 + client_id_length;
  if(username_length > 0)
    packetLen = packetLen + 2 + username_length;
  if(password_length > 0)
    packetLen = packetLen+ 2 + password_length;
  mqtt_message[0] = 16;       //0x10 // MQTT Message Type CONNECT
  mqtt_message[1] = packetLen - 2;  //Ê£Ó೤¶È(²»°üÀ¨¹Ì¶¨Í·²¿)
  baseIndex = 2;
  if(packetLen > 127)
  {
    mqtt_message[2] = 1;      //packetLen/127;   
    baseIndex = 3;
  }
  mqtt_message[baseIndex++] = 0;    // Protocol Name Length MSB   
  mqtt_message[baseIndex++] = 4;    // Protocol Name Length LSB   
  mqtt_message[baseIndex++] = 77;   // ASCII Code for M   
  mqtt_message[baseIndex++] = 81;   // ASCII Code for Q   
  mqtt_message[baseIndex++] = 84;   // ASCII Code for T   
  mqtt_message[baseIndex++] = 84;   // ASCII Code for T   
  mqtt_message[baseIndex++] = 4;    // MQTT Protocol version = 4   
  mqtt_message[baseIndex++] = 194;    // conn flags
  mqtt_message[baseIndex++] = 0;    // Keep-alive Time Length MSB   
  mqtt_message[baseIndex++] = 300;   // Keep-alive Time Length LSB   
  mqtt_message[baseIndex++] = (0xff00&client_id_length)>>8;// Client ID length MSB
  mqtt_message[baseIndex++] = 0xff&client_id_length;  // Client ID length LSB
  // Client ID
  for(i = 0; i < client_id_length; i++)
  {
    mqtt_message[baseIndex + i] = client_id;   
  }
  baseIndex = baseIndex+client_id_length;


  if(username_length > 0)
  {
    //username   
    mqtt_message[baseIndex++] = (0xff00&username_length)>>8;//username length MSB   
    mqtt_message[baseIndex++] = 0xff&username_length; //username length LSB   
    for(i = 0; i < username_length ; i++)
    {
      mqtt_message[baseIndex + i] = username;   
    }
    baseIndex = baseIndex + username_length;
  }

  if(password_length > 0)
  {
    //password   
    mqtt_message[baseIndex++] = (0xff00&password_length)>>8;//password length MSB   
    mqtt_message[baseIndex++] = 0xff&password_length; //password length LSB   
    for(i = 0; i < password_length ; i++)
    {
      mqtt_message[baseIndex + i] = password;   
    }
    baseIndex += password_length;
  }

  return baseIndex;   
}
/*********MQTT发布消息包******************************/
u8 mqtt_publish_message(u8 *mqtt_message, char * topic, char * message, u8 qos)
{  
  u16 topic_length = strlen(topic);   
  u16 message_length = strlen(message);  
  u16 i,index=0;  
  static u16 id=0;

  mqtt_message[index++] = 48; //0x30 // MQTT Message Type PUBLISH   
  if(qos)
    mqtt_message[index++] = 2 + topic_length + 2 + message_length;
  else
    mqtt_message[index++] = 2 + topic_length + message_length;   // Remaining length   
  mqtt_message[index++] = (0xff00&topic_length)>>8;
  mqtt_message[index++] = 0xff&topic_length;

  // Topic   
  for(i = 0; i < topic_length; i++)
  {
    mqtt_message[index + i] = topic;
  }
  index += topic_length;

  if(qos)
  {
    mqtt_message[index++] = (0xff00&id)>>8;
    mqtt_message[index++] = 0xff&id;
    id++;
  }

  // Message
  for(i = 0; i < message_length; i++)
  {
    mqtt_message[index + i] = message;
  }
  index += message_length;
  return index;
}
/*****************发布确认包****************************/
u8 mqtt_puback_message(u8 *mqtt_message)
{
  static u16 id=0;
  mqtt_message[0] = 64;       //0x40 //ÏûÏ¢ÀàÐͺͱêÖ¾ PUBACK
  mqtt_message[1] = 2;        //Ê£Ó೤¶È(²»°üÀ¨¹Ì¶¨Í·²¿)
  mqtt_message[2] = (0xff00&id)>>8; //ÏûÏ¢±êʶ·û
  mqtt_message[3] = 0xff&id;      //ÏûÏ¢±êʶ·û
  id++;
  return 4;
}
/*********构建订阅请求*********/
u16 mqtt_subscribe_message(u8 *mqtt_message,char *topic,u8 qos,u8 whether)
{   
  u16 topic_len = strlen(topic);
  u16 i,index = 0;
  static u16 id=0;

  id++;
  if(whether)
    mqtt_message[index++] = 130;        
  else
    mqtt_message[index++] = 162;        
  mqtt_message[index++] = topic_len + 5;     
  mqtt_message[index++] = (0xff00&id)>>8;   
  mqtt_message[index++] = 0xff&id;      
  mqtt_message[index++] = (0xff00&topic_len)>>8;  
  mqtt_message[index++] = 0xff&topic_len;     
  for (i = 0;i < topic_len; i++)
  {
    mqtt_message[index + i] = topic;
  }
  index += topic_len;

  if(whether)
  {
    mqtt_message[index] = qos;//QoS¼¶±ð
    index++;
  }
  return index;
}
/***********构建MQTT请求PING包********************/
u8 mqtt_ping_message(u8 *mqtt_message)
{
  mqtt_message[0] = 192;  //0xC0 //ÏûÏ¢ÀàÐͺͱêÖ¾ PING
  mqtt_message[1] = 0;  //Ê£Ó೤¶È(²»°üÀ¨¹Ì¶¨Í·²¿)
  return 2;
}
/************构建断开包******************/
u8 mqtt_disconnect_message(u8 *mqtt_message)
{
  mqtt_message[0] = 224;  //0xE0 //ÏûÏ¢ÀàÐͺͱêÖ¾ DISCONNECT
  mqtt_message[1] = 0;  //Ê£Ó೤¶È(²»°üÀ¨¹Ì¶¨Í·²¿)
  return 2;
}

/***************单路连接发送数据****************/
u8 ESP_SendTcp(u8 *data,u16 len)
{
  while(debug.read()>= 0){}
    debug.listen();
   debug.print("AT+CIPSEND=");
   debug.println(len);
   Serial.println("len:"+len);
   Serial.println("...ing2................");
   delay(1000);
    if(debug.find( ">" ) )
    {
      Serial.println("is   sendind  >>>>");
      String mqtt_data="";
      u16 i;
       for(i=0;i<len;i++)
      {
        //mqtt_data +=data;
        debug.write(data);
        Serial.write(data);
        }
         debug.listen();
         debug.println();
         delay(500);
         //debug.println();
    }
    else
    {

      Serial.println("send  error");
        return false;
    }
    if( debug.find("OK") )
    {
      Serial.println("send succfful!!");
       return true;
    }
    else
    {
      Serial.println("send  wrong");
       return false;
    }
}

boolean mqtt_connect(){
  Serial.println("is  connect mqtt");
  u16 len,i;
  String cmd = "AT+CIPSTART=\"TCP\",\"";
    cmd += IP_mqtt;
    cmd += "\",61613";
    while(debug.read()>= 0){}
    debug.listen();
    sendDebug(cmd);
    delay(1500);
    while( !debug.find( "OK" ) )
    {
        Serial.print( "." );
        sendDebug(cmd);
        //return;
    }
     while(debug.read()>= 0){}
     debug.listen();
  len=mqtt_connect_message(mqtt_msg,"123456","admin","password");//id=123456,用户名和密码''
  ESP_SendTcp(mqtt_msg,len);
  while(!debug.find("P")){
    Serial.print("ljing...");
    }
Serial.print( "Begin to descriped..." );
len=mqtt_subscribe_message(mqtt_msg,des,0,1);

Serial.println("...ing................");
if(ESP_SendTcp(mqtt_msg,len))
    Serial.print("dingyue  sucessfull");
    delay(1000);
    while(debug.read()>= 0){}
  return isconnect;
}
boolean connectWiFi_mqtt(){
    // 连接到Wifi
     while(Serial.read()>= 0){}
    while(debug.read()>= 0){}
    debug.println("AT+RST");
    delay(1000);
    while(debug.read()>= 0){}
    debug.listen();
    debug.println("AT+CWMODE=1");
    delay(500);
    String cmd="AT+CWJAP=\"";
    cmd+=SSID;
    cmd+="\",\"";
    cmd+=PASS;
    cmd+="\"";
    while(debug.read()>= 0){}
    debug.listen();
    sendDebug(cmd);
    delay(3000);
    while(!debug.find("OK"))
    {
        Serial.println(".");
       // return true;
    }
     cmd = "AT+CIPMUX=0";
     while(debug.read()>= 0){}
     debug.listen();
     sendDebug( cmd );
      if(debug.find("OK"))
    {
        Serial.println("one   OK");
       // return true;
    }
    else if( debug.find( "Error") )
    {
        Serial.print( "RECEIVED: Error" );
        return false;
    }
}
  void updateDHT11_mqtt( String T, String H )
{
   Serial.println("is going to send DHT11  1");
  while(debug.read()>= 0){}
      debug.listen();  
    // 使用AT指令上传
    Serial.println("is going to send DHT11  2");
    delay(1000);
    String cmd="";
    u16 len;
    u16 i;
    cmd ="Temperature=!"+T+"!Humidity=!"+H+"!%"+ "Qiti=!"+cal+"!\r\n" ;
    char mg[100];
    strcpy(mg,cmd.c_str());
     Serial.print( "Begin send chuan gan qi" );
    len=mqtt_publish_message(mqtt_msg,pub,mg,0);
     if(ESP_SendTcp(mqtt_msg,len))
    Serial.print("send data sucessfull");
}
void send_mqtt_dht(){
        ReadDHT();
       String TT,HH;
        HH = String(dht_dat[0])+'.'+String(dht_dat[1]);
        TT = String(dht_dat[2])+'.'+String(dht_dat[3]);
        cal=analogRead(qiti);//读取烟雾传感器数值
        Serial.println(cal);//输出模拟值，并将其打印出来
        updateDHT11_mqtt( TT, HH );
        while(debug.read()>= 0){}
        debug.listen();  
}
void mode_mqtt(){
  connectWiFi_mqtt();
  while(Serial.read()>= 0){}
  while(debug.read()>= 0){}
  delay(700);
  mqtt_connect();
  send_mqtt_dht();
  while(MODE==1){//一直处于此循环
  send_mqtt_dht();
  delay(1500);//每1.5秒发送一次(粗略)
    keepAlive();
  debug.listen();
  /*if(debug.find('P')){
   //这里写接收的逻辑，根据实际情况而定
  }*/

}
}

/****************************************模式选择**********************************************/
void SELECT_MODE()
{
  switch(MODE){
    case 1:
    mode_mqtt();
    break;

    case 2:
    break;

    case 3:
    break;

    case 0:
    break;
    }
}
void  keepAlive(){
           u8 ping[2];
              ping[0]=0xc0;
              ping[1]=0x00;
              if(ESP_SendTcp(ping,2))//·心跳包
              {
                while(debug.read()>= 0){}
                Serial.println("MQTT_Ping ..Keep Alive.");
             }
             is_send=false;
}
void event(){
  switch(MODE)
  {
    case 1:
   keepAlive();
    break;

    case 2:
    break;

    case 3:

    break;
    case 0:

    break;
    }
}
void start_timer(int t){
    /*FlexiTimer2::set(t,1.0/1000,event);
    FlexiTimer2::start();*/
}
void stop_timer(){
  //FlexiTimer2::stop();
}


[/mw_shl_code]



我时间有限。，暂时不提供基础教程。我也是从零开始的，自己一步步看别人的教程，然后自己跟着做，只要肯下功夫是肯定没问题的，

esp_mqtt.zip

4.15 KB, 下载次数: 134
回复

举报

jiangzhiwei
 楼主| 发表于 2018-12-6 14:40 | 显示全部楼层
图片怎么变附件了。。。我明明是加在字后面的呀
回复 支持 反对

举报

qq957606954
发表于 2019-1-16 11:48 | 显示全部楼层
为什么我的mqtt一直在链接，没有链接成功的提示
回复 支持 反对

举报

jiangzhiwei
 楼主| 发表于 2019-1-21 14:49 | 显示全部楼层
qq957606954 发表于 2019-1-16 11:48
为什么我的mqtt一直在链接，没有链接成功的提示

检查云服务器的入站规则  是否有添加你的mqtt设置的端口
回复 支持 反对

举报

BIBOBIBO
发表于 2019-4-7 15:17 | 显示全部楼层
请问阿里云ecs服务器实例安全组入站规则中的授权对象应该怎么填什么？
回复 支持 反对

举报

tiankefeng567
发表于 2019-6-24 23:52 | 显示全部楼层
将的很详细，学习了，谢谢楼主
回复 支持 反对

举报

tangqishun
发表于 2020-3-15 15:37 | 显示全部楼层
很厉害，源码是哪个库的？？
回复 支持 反对

举报

返回列表 发新帖
高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

小黑屋|Archiver|手机版|Arduino中文社区

GMT+8, 2024-11-28 06:32 , Processed in 0.270639 second(s), 18 queries .

Powered by Discuz! X3.4

Copyright © 2001-2021, Tencent Cloud.

快速回复 返回顶部 返回列表