|
|
本帖最后由 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](data/attachment/forum/201812/06/134026stzmtyeoa4ohexsu.png "Y79AZ{1LI9Q7DP[(][KOIIA.png")
购买之后右上角进入控制台管理你的云服务器
进入点击 右侧的 “实例”管理 云服务器 ，
![U$]JUQ%AXH(HD{K%I6I}UH3.png](data/attachment/forum/201812/06/133904qtbh9bxnpwwyhzxp.png "U$]JUQ%AXH(HD{K%I6I}UH3.png")
得到你的公网ip，
![6E9RM@5{$K]JY`WRO[RY{0V.png](data/attachment/forum/201812/06/134833vdqfhs02hfqfxfjh.png "6E9RM@5{$K]JY`WRO[RY{0V.png")
，然后远程桌面连接
/****下面来介绍搭建MQTT服务器**********************/
http://activemq.apache.org/apollo/download.html 进入下载
根据自己购买的服务的系统选择版本类型。我的是windows ，所以选择第二个，下载下载好后解压
注意 在配置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文件夹
之后再输入 apollo create mybroker 即创建一个名为mybroker 的文件夹
创建成功会出现以下提示并多了一个mybroker的文件夹
之后 用命令行进入该文件夹 下的 bin 目录 ，
最后运行apollo-broker.cmd 出现如下
![7HECQ4]7I7)}_H5IV%@}9Y7.png](data/attachment/forum/201812/06/141308fn2n3riyqirstlpv.png "7HECQ4]7I7)}_H5IV%@}9Y7.png")
![G`OL5(5P]Q5]EVNF1SJKB}W.png](data/attachment/forum/201812/06/141323aojlu8828i42shsb.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](data/attachment/forum/201812/06/141722azp8zb66h76b6n41.png "@)T64@{Z9%$P$TXWJ]STN03.png")
最后连接测试成功
![H978%]D6UF[EC5GC$RRSNU0.png](data/attachment/forum/201812/06/142140h8kzc285t18vc48z.png "H978%]D6UF[EC5GC$RRSNU0.png")
/********************************下面我来贴上我的AT指令开发MQTT的代码*******************************************************/
不过一切前提是你能建立TCP连接
主要分为2大块
一，建立TCP客户端连接
二，MQTT连接包封装函数 这是我从找到的32 源码的上面移植的封装函数，连接包的格式上有一个地方小错误，这边已经改正了，测试能正常连接
这边传感器没有接线上去，但是数据传输成功了！
[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]
我时间有限。，暂时不提供基础教程。我也是从零开始的，自己一步步看别人的教程，然后自己跟着做，只要肯下功夫是肯定没问题的，
|
|
发表于 2018-12-6 14:34
楼主