先给你们看个最终代码结构吧,因为我改过的代码会加密,所以我只能放一部分源码,另外一部分源码我会直接贴在博客,具体使用我会在博客里面说明!

 1.MQTTPacket源码库(MQTTPacket源码地址)

 2.MQTTClient.c

/******************************************************************************** Copyright (c) 2014 IBM Corp.** All rights reserved. This program and the accompanying materials* are made available under the terms of the Eclipse Public License v1.0* and Eclipse Distribution License v1.0 which accompany this distribution.** The Eclipse Public License is available at*    http://www.eclipse.org/legal/epl-v10.html* and the Eclipse Distribution License is available at*   http://www.eclipse.org/org/documents/edl-v10.php.** Contributors:*    Allan Stockdill-Mander/Ian Craggs - initial API and implementation and/or initial documentation*******************************************************************************/#include "MQTTClient.h"void NewMessageData(MessageData* md, MQTTString* aTopicName, MQTTMessage* aMessgage) {md->topicName = aTopicName;md->message = aMessgage;
}int getNextPacketId(Client *c) {return c->next_packetid = (c->next_packetid == MAX_PACKET_ID) ? 1 : c->next_packetid + 1;
}int sendPacket(Client* c, int length, Timer* timer)
{int rc = FAILURE, sent = 0;while (sent < length && !expired(timer)){rc = c->ipstack->mqttwrite(c->ipstack, &c->buf[sent], length, left_ms(timer));if (rc < 0)  // there was an error writing the databreak;sent += rc;}if (sent == length){countdown(&c->ping_timer, c->keepAliveInterval); // record the fact that we have successfully sent the packet    rc = SUCCESS;}elserc = FAILURE;return rc;
}void MQTTClient(Client* c, Network* network, unsigned int command_timeout_ms, unsigned char* buf, size_t buf_size, unsigned char* readbuf, size_t readbuf_size)
{int i;c->ipstack = network;for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i)c->messageHandlers[i].topicFilter = 0;c->command_timeout_ms = command_timeout_ms;c->buf = buf;c->buf_size = buf_size;c->readbuf = readbuf;c->readbuf_size = readbuf_size;c->isconnected = 0;c->ping_outstanding = 0;c->defaultMessageHandler = NULL;InitTimer(&c->ping_timer);
}int decodePacket(Client* c, int* value, int timeout)
{unsigned char i;int multiplier = 1;int len = 0;const int MAX_NO_OF_REMAINING_LENGTH_BYTES = 4;*value = 0;do{int rc = MQTTPACKET_READ_ERROR;if (++len > MAX_NO_OF_REMAINING_LENGTH_BYTES){rc = MQTTPACKET_READ_ERROR; /* bad data */goto exit;}rc = c->ipstack->mqttread(c->ipstack, &i, 1, timeout);if (rc != 1)goto exit;*value += (i & 127) * multiplier;multiplier *= 128;} while ((i & 128) != 0);
exit:return len;
}int readPacket(Client* c, Timer* timer) 
{int rc = FAILURE;MQTTHeader header = {0};int len = 0;int rem_len = 0;/* 1. read the header byte.  This has the packet type in it */if ((rc = c->ipstack->mqttread(c->ipstack, c->readbuf, 1, left_ms(timer))) != 1)goto exit;len = 1;/* 2. read the remaining length.  This is variable in itself */decodePacket(c, &rem_len, left_ms(timer));len += MQTTPacket_encode(c->readbuf + 1, rem_len); /* put the original remaining length back into the buffer *//* 3. read the rest of the buffer using a callback to supply the rest of the data */if (rem_len > 0 && ((rc = c->ipstack->mqttread(c->ipstack, c->readbuf + len, rem_len, left_ms(timer)) )!= rem_len))goto exit;header.byte = c->readbuf[0];rc = header.bits.type;
exit:return rc;
}// assume topic filter and name is in correct format
// # can only be at end
// + and # can only be next to separator
char isTopicMatched(char* topicFilter, MQTTString* topicName)
{char* curf = topicFilter;char* curn = topicName->lenstring.data;char* curn_end = curn + topicName->lenstring.len;while (*curf && curn < curn_end){if (*curn == '/' && *curf != '/')break;if (*curf != '+' && *curf != '#' && *curf != *curn)break;if (*curf == '+'){   // skip until we meet the next separator, or end of stringchar* nextpos = curn + 1;while (nextpos < curn_end && *nextpos != '/')nextpos = ++curn + 1;}else if (*curf == '#')curn = curn_end - 1;    // skip until end of stringcurf++;curn++;};return (curn == curn_end) && (*curf == '\0');
}int deliverMessage(Client* c, MQTTString* topicName, MQTTMessage* message)
{int i;int rc = FAILURE;// we have to find the right message handler - indexed by topicfor (i = 0; i < MAX_MESSAGE_HANDLERS; ++i){if (c->messageHandlers[i].topicFilter != 0 && (MQTTPacket_equals(topicName, (char*)c->messageHandlers[i].topicFilter) ||isTopicMatched((char*)c->messageHandlers[i].topicFilter, topicName))){if (c->messageHandlers[i].fp != NULL){MessageData md;NewMessageData(&md, topicName, message);c->messageHandlers[i].fp(&md);rc = SUCCESS;}}}if (rc == FAILURE && c->defaultMessageHandler != NULL) {MessageData md;NewMessageData(&md, topicName, message);c->defaultMessageHandler(&md);rc = SUCCESS;}   return rc;
}int keepalive(Client* c)
{int rc = SUCCESS;if (c->keepAliveInterval == 0){rc = SUCCESS;goto exit;}if (expired(&c->ping_timer)){if (!c->ping_outstanding){Timer timer;InitTimer(&timer);countdown_ms(&timer, 1000);int len = MQTTSerialize_pingreq(c->buf, c->buf_size);if (len > 0 && (rc = sendPacket(c, len, &timer)) == SUCCESS) // send the ping packetc->ping_outstanding = 1;}else{++(c->fail_count);if (c->fail_count >= MAX_FAIL_ALLOWED){rc = DISCONNECTED;goto exit;}}countdown(&(c->ping_timer), c->keepAliveInterval);}exit:return rc;
}int cycle(Client* c, Timer* timer)
{int len = 0,rc = SUCCESS;// read the socket, see what work is dueunsigned short packet_type = readPacket(c, timer);if(packet_type <= 0){//printf("[%s][%d]readPacket retrun socket close \n",__FUNCTION__, __LINE__);//printf("[%s]....packet_type = %d\n", __FUNCTION__, packet_type);rc =  DISCONNECTED;goto exit;}switch (packet_type){case CONNACK:case PUBACK:case SUBACK:break;case PUBLISH:{MQTTString topicName;MQTTMessage msg;if (MQTTDeserialize_publish((unsigned char*)&msg.dup, (int*)&msg.qos, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,(unsigned char**)&msg.payload, (int*)&msg.payloadlen, c->readbuf, c->readbuf_size) != 1)goto exit;deliverMessage(c, &topicName, &msg);if (msg.qos != QOS0){if (msg.qos == QOS1)len = MQTTSerialize_ack(c->buf, c->buf_size, PUBACK, 0, msg.id);else if (msg.qos == QOS2)len = MQTTSerialize_ack(c->buf, c->buf_size, PUBREC, 0, msg.id);if (len <= 0)rc = FAILURE;elserc = sendPacket(c, len, timer);if (rc == FAILURE){goto exit; // there was a problem}}break;}case PUBREC:{unsigned short mypacketid;unsigned char dup, type;if (MQTTDeserialize_ack(&type, &dup, &mypacketid, c->readbuf, c->readbuf_size) != 1)rc = FAILURE;else if ((len = MQTTSerialize_ack(c->buf, c->buf_size, PUBREL, 0, mypacketid)) <= 0)rc = FAILURE;else if ((rc = sendPacket(c, len, timer)) != SUCCESS) // send the PUBREL packetrc = FAILURE; // there was a problemif (rc == FAILURE){goto exit; // there was a problem}break;}case PUBCOMP:break;case PINGRESP:c->ping_outstanding = 0;c->fail_count = 0;break;}if (c->isconnected)rc = keepalive(c);
exit:if (rc == SUCCESS)rc = packet_type;;return rc;
}int MQTTYield(Client* c, int timeout_ms)
{int rc = SUCCESS;Timer timer;InitTimer(&timer);    countdown_ms(&timer, timeout_ms);while (!expired(&timer)){rc = cycle(c, &timer);if (rc == DISCONNECTED){break;}rc = SUCCESS;}return rc;
}// only used in single-threaded mode where one command at a time is in process
int waitfor(Client* c, int packet_type, Timer* timer)
{int rc = FAILURE;do{if (expired(timer)) break; // we timed out}while ((rc = cycle(c, timer)) != packet_type);  return rc;
}int MQTTConnect(Client* c, MQTTPacket_connectData* options)
{Timer connect_timer;int rc = FAILURE;MQTTPacket_connectData default_options = MQTTPacket_connectData_initializer;int len = 0;InitTimer(&connect_timer);countdown_ms(&connect_timer, c->command_timeout_ms);if (c->isconnected) // don't send connect packet again if we are already connectedgoto exit;if (options == 0)options = &default_options; // set default options if none were suppliedc->keepAliveInterval = options->keepAliveInterval;countdown(&c->ping_timer, c->keepAliveInterval);printf("[%s]c->keepAliveInterval = %d\n", __FUNCTION__,c->keepAliveInterval);if ((len = MQTTSerialize_connect(c->buf, c->buf_size, options)) <= 0)goto exit;if ((rc = sendPacket(c, len, &connect_timer)) != SUCCESS)  // send the connect packetgoto exit; // there was a problem// this will be a blocking call, wait for the connackif (waitfor(c, CONNACK, &connect_timer) == CONNACK){unsigned char connack_rc = 255;char sessionPresent = 0;if (MQTTDeserialize_connack((unsigned char*)&sessionPresent, &connack_rc, c->readbuf, c->readbuf_size) == 1)rc = connack_rc;elserc = FAILURE;}elserc = FAILURE;exit:if (rc == SUCCESS)c->isconnected = 1;return rc;
}int MQTTSubscribe(Client* c, const char* topicFilter, enum QoS qos, messageHandler messageHandler)
{ int rc = FAILURE;  Timer timer;int len = 0;MQTTString topic = MQTTString_initializer;topic.cstring = (char *)topicFilter;InitTimer(&timer);countdown_ms(&timer, c->command_timeout_ms);if (!c->isconnected)goto exit;len = MQTTSerialize_subscribe(c->buf, c->buf_size, 0, getNextPacketId(c), 1, &topic, (int*)&qos);if (len <= 0)goto exit;if ((rc = sendPacket(c, len, &timer)) != SUCCESS) // send the subscribe packetgoto exit;             // there was a problemif (waitfor(c, SUBACK, &timer) == SUBACK)      // wait for suback {int count = 0, grantedQoS = -1;unsigned short mypacketid;if (MQTTDeserialize_suback(&mypacketid, 1, &count, &grantedQoS, c->readbuf, c->readbuf_size) == 1)rc = grantedQoS; // 0, 1, 2 or 0x80 if (rc != 0x80){int i;for (i = 0; i < MAX_MESSAGE_HANDLERS; ++i){if (c->messageHandlers[i].topicFilter == 0){c->messageHandlers[i].topicFilter = topicFilter;c->messageHandlers[i].fp = messageHandler;rc = 0;break;}}}}else rc = FAILURE;exit:return rc;
}int MQTTUnsubscribe(Client* c, const char* topicFilter)
{   int rc = FAILURE;Timer timer;    MQTTString topic = MQTTString_initializer;topic.cstring = (char *)topicFilter;int len = 0;InitTimer(&timer);countdown_ms(&timer, c->command_timeout_ms);if (!c->isconnected)goto exit;if ((len = MQTTSerialize_unsubscribe(c->buf, c->buf_size, 0, getNextPacketId(c), 1, &topic)) <= 0)goto exit;if ((rc = sendPacket(c, len, &timer)) != SUCCESS) // send the subscribe packetgoto exit; // there was a problemif (waitfor(c, UNSUBACK, &timer) == UNSUBACK){unsigned short mypacketid;  // should be the same as the packetid aboveif (MQTTDeserialize_unsuback(&mypacketid, c->readbuf, c->readbuf_size) == 1)rc = 0; }elserc = FAILURE;exit:return rc;
}int MQTTPublish(Client* c, const char* topicName, MQTTMessage* message)
{int rc = FAILURE;Timer timer;   MQTTString topic = MQTTString_initializer;topic.cstring = (char *)topicName;int len = 0;InitTimer(&timer);countdown_ms(&timer, c->command_timeout_ms);if (!c->isconnected)goto exit;if (message->qos == QOS1 || message->qos == QOS2)message->id = getNextPacketId(c);len = MQTTSerialize_publish(c->buf, c->buf_size, 0, message->qos, message->retained, message->id, topic, (unsigned char*)message->payload, message->payloadlen);if (len <= 0)goto exit;if ((rc = sendPacket(c, len, &timer)) != SUCCESS) // send the subscribe packetgoto exit; // there was a problemif (message->qos == QOS1){if (waitfor(c, PUBACK, &timer) == PUBACK){unsigned short mypacketid;unsigned char dup, type;if (MQTTDeserialize_ack(&type, &dup, &mypacketid, c->readbuf, c->readbuf_size) != 1)rc = FAILURE;}elserc = FAILURE;}else if (message->qos == QOS2){if (waitfor(c, PUBCOMP, &timer) == PUBCOMP){unsigned short mypacketid;unsigned char dup, type;if (MQTTDeserialize_ack(&type, &dup, &mypacketid, c->readbuf, c->readbuf_size) != 1)rc = FAILURE;}elserc = FAILURE;}exit:return rc;
}int MQTTDisconnect(Client* c)
{  int rc = FAILURE;Timer timer;     // we might wait for incomplete incoming publishes to completeint len = MQTTSerialize_disconnect(c->buf, c->buf_size);InitTimer(&timer);countdown_ms(&timer, c->command_timeout_ms);if (len > 0)rc = sendPacket(c, len, &timer);            // send the disconnect packetc->isconnected = 0;return rc;
}

 3.MQTTClient.h

/******************************************************************************** Copyright (c) 2014 IBM Corp.** All rights reserved. This program and the accompanying materials* are made available under the terms of the Eclipse Public License v1.0* and Eclipse Distribution License v1.0 which accompany this distribution.** The Eclipse Public License is available at*    http://www.eclipse.org/legal/epl-v10.html* and the Eclipse Distribution License is available at*   http://www.eclipse.org/org/documents/edl-v10.php.** Contributors:*    Allan Stockdill-Mander/Ian Craggs - initial API and implementation and/or initial documentation*******************************************************************************/#ifndef __MQTT_CLIENT_C_
#define __MQTT_CLIENT_C_#include "MQTTPacket/MQTTPacket.h"
#include "stdio.h"
#include "MQTTLinux.h" #define MAX_PACKET_ID 65535
#define MAX_MESSAGE_HANDLERS 5
#define MAX_FAIL_ALLOWED  2enum QoS { QOS0, QOS1, QOS2 };// all failure return codes must be negative
enum returnCode {DISCONNECTED = -3,  BUFFER_OVERFLOW = -2, FAILURE = -1, SUCCESS = 0 };void NewTimer(Timer*);typedef struct MQTTMessage MQTTMessage;typedef struct MessageData MessageData;struct MQTTMessage
{enum QoS qos;char retained;char dup;unsigned short id;void *payload;size_t payloadlen;
};struct MessageData
{MQTTMessage* message;MQTTString* topicName;
};typedef void (*messageHandler)(MessageData*);typedef struct Client Client;int MQTTConnect (Client*, MQTTPacket_connectData*);
int MQTTPublish (Client*, const char*, MQTTMessage*);
int MQTTSubscribe (Client*, const char*, enum QoS, messageHandler);
int MQTTUnsubscribe (Client*, const char*);
int MQTTDisconnect (Client*);
int MQTTYield (Client*, int);void setDefaultMessageHandler(Client*, messageHandler);void MQTTClient(Client*, Network*, unsigned int, unsigned char*, size_t, unsigned char*, size_t);struct Client {unsigned int next_packetid;unsigned int command_timeout_ms;size_t buf_size, readbuf_size;unsigned char *buf;  unsigned char *readbuf; unsigned int keepAliveInterval;char ping_outstanding;int fail_count;int isconnected;struct MessageHandlers{const char* topicFilter;void (*fp) (MessageData*);} messageHandlers[MAX_MESSAGE_HANDLERS];      // Message handlers are indexed by subscription topicvoid (*defaultMessageHandler) (MessageData*);Network* ipstack;Timer ping_timer;
};#define DefaultClient {0, 0, 0, 0, NULL, NULL, 0, 0, 0}#endif

 4.MQTTLinux.c

/******************************************************************************** Copyright (c) 2014 IBM Corp.** All rights reserved. This program and the accompanying materials* are made available under the terms of the Eclipse Public License v1.0* and Eclipse Distribution License v1.0 which accompany this distribution.** The Eclipse Public License is available at*    http://www.eclipse.org/legal/epl-v10.html* and the Eclipse Distribution License is available at*   http://www.eclipse.org/org/documents/edl-v10.php.** Contributors:*    Allan Stockdill-Mander - initial API and implementation and/or initial documentation*******************************************************************************/
// #include 
// #include 
#include "MQTTLinux.h"
//#include "global.h"
//#include "DC_iot_port.h"char expired(Timer* timer)
{struct timeval now, res;gettimeofday(&now, NULL);timersub(&timer->end_time, &now, &res);		return res.tv_sec < 0 || (res.tv_sec == 0 && res.tv_usec <= 0);
}void countdown_ms(Timer* timer, unsigned int timeout)
{struct timeval now;gettimeofday(&now, NULL);struct timeval interval = {timeout / 1000, (timeout % 1000) * 1000};timeradd(&now, &interval, &timer->end_time);
}void countdown(Timer* timer, unsigned int timeout)
{struct timeval now;gettimeofday(&now, NULL);struct timeval interval = {timeout, 0};timeradd(&now, &interval, &timer->end_time);
}int left_ms(Timer* timer)
{struct timeval now, res;gettimeofday(&now, NULL);timersub(&timer->end_time, &now, &res);//printf("left %d ms\n", (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000);return (res.tv_sec < 0) ? 0 : res.tv_sec * 1000 + res.tv_usec / 1000;
}void InitTimer(Timer* timer)
{timer->end_time = (struct timeval){0, 0};
}int linux_read(Network* n, unsigned char* buffer, int len, int timeout_ms)
{struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000};//mDEBUG("[%s]timeout_ms = %d \n", __FUNCTION__, timeout_ms);if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0)){interval.tv_sec = 0;interval.tv_usec = 100;}setsockopt(n->my_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&interval, sizeof(struct timeval));int bytes = 0;while (bytes < len){int rc = recv(n->my_socket, &buffer[bytes], (size_t)(len - bytes), 0);//mDEBUG("[%s]socket recv rc = %d \n", __FUNCTION__, rc);if(rc == 0){//mDEBUG("[%s]socket close \n", __FUNCTION__);bytes = 0;break;}if (rc == -1){if (errno != ENOTCONN && errno != ECONNRESET){bytes = -1;break;}}elsebytes += rc;}return bytes;
}int linux_write(Network* n, unsigned char* buffer, int len, int timeout_ms)
{struct timeval tv;tv.tv_sec = 0;  /* 30 Secs Timeout */tv.tv_usec = timeout_ms * 1000;  // Not init'ing this can cause strange errorssetsockopt(n->my_socket, SOL_SOCKET, SO_RCVTIMEO, (char *)&tv,sizeof(struct timeval));int	rc = write(n->my_socket, buffer, len);return rc;
}void linux_disconnect(Network* n)
{close(n->my_socket);}void NewNetwork(Network* n)
{n->my_socket = 0;n->mqttread = linux_read;n->mqttwrite = linux_write;n->disconnect = linux_disconnect;
}int ConnectNetwork(Network* n, char* addr, int port)
{int type = SOCK_STREAM;struct sockaddr_in address;int rc = -1;sa_family_t family = AF_INET;struct addrinfo *result = NULL;struct addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM, IPPROTO_TCP, 0, NULL, NULL, NULL};if ((rc = getaddrinfo(addr, NULL, &hints, &result)) == 0){struct addrinfo* res = result;/* prefer ip4 addresses */while (res){if (res->ai_family == AF_INET){result = res;break;}res = res->ai_next;}if (result->ai_family == AF_INET){address.sin_port = htons(port);address.sin_family = family = AF_INET;address.sin_addr = ((struct sockaddr_in*)(result->ai_addr))->sin_addr;}elserc = -1;freeaddrinfo(result);}if (rc == 0){n->my_socket = socket(family, type, 0);if (n->my_socket != -1){int opt = 1;			rc = connect(n->my_socket, (struct sockaddr*)&address, sizeof(address));}}return rc;
}

5.MQTTLinux.h 

/******************************************************************************** Copyright (c) 2014 IBM Corp.** All rights reserved. This program and the accompanying materials* are made available under the terms of the Eclipse Public License v1.0* and Eclipse Distribution License v1.0 which accompany this distribution.** The Eclipse Public License is available at*    http://www.eclipse.org/legal/epl-v10.html* and the Eclipse Distribution License is available at*   http://www.eclipse.org/org/documents/edl-v10.php.** Contributors:*    Allan Stockdill-Mander - initial API and implementation and/or initial documentation*******************************************************************************/#ifndef __MQTT_LINUX_
#define __MQTT_LINUX_#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include 
#include #include 
#include 
#include // #include 
// #include 
//#include "DC_iot_port.h"typedef struct Timer Timer;struct Timer {struct timeval end_time;
};typedef struct Network Network;struct Network
{#if IOT_SSL_ENABLESSL_CTX *ctx;SSL *ssl;
#endifint my_socket;int (*mqttread) (Network*, unsigned char*, int, int);int (*mqttwrite) (Network*, unsigned char*, int, int);void (*disconnect) (Network*);
};char expired(Timer*);
void countdown_ms(Timer*, unsigned int);
void countdown(Timer*, unsigned int);
int left_ms(Timer*);void InitTimer(Timer*);int linux_read(Network*, unsigned char*, int, int);
int linux_write(Network*, unsigned char*, int, int);
void linux_disconnect(Network*);
void NewNetwork(Network*);int ConnectNetwork(Network*, char*, int);#endif

 6.mqtt.c

#include #include "MQTTClient.h"
#include "mqtt.h"
#include "pthread.h"
#include "string.h"
#include "unistd.h"
#include "sys/stat.h"
#include "sys/types.h"
#include "sys/socket.h"
#include "netinet/in.h"
#include "arpa/inet.h"
#include "fcntl.h"#define MQTT_TOPIC_SIZE     (128)		//订阅和发布主题长度
#define MQTT_BUF_SIZE       (8 * 1024) 	//接收后发送缓冲区大小#define MQTT_HOST "114.114.114"		//ip地址
#define MQTT_PORT 1883					//端口号
#define MQTT_USER "admin"				//用户名
#define MQTT_PASS "password"			//密码
#define MQTT_CLIENT_ID "admin"		//客户端标识
#define COMMAND_TIMEOUT_MS 10000    //连接超时时间 如果mqtt服务器响应比较慢的话,可以把这个时间拉长typedef struct {Network Network;Client Client;char sub_topic[MQTT_TOPIC_SIZE];		//存放订阅主题char pub_topic[MQTT_TOPIC_SIZE];		//存放发布主题char mqtt_buffer[MQTT_BUF_SIZE];		//发送缓冲区char mqtt_read_buffer[MQTT_BUF_SIZE];	//接收缓冲区unsigned char willFlag;					MQTTPacket_willOptions will;char will_topic[MQTT_TOPIC_SIZE];		//存放遗嘱主题pMessageArrived_Fun DataArrived_Cb;
}Cloud_MQTT_t;typedef struct{enum iot_ctrl_status_t iotstatus;char model[7];char company[32];
} iot_device_info_t;//主题结构体struct opts_struct {char    *clientid;int     nodelimiter;char    *delimiter;enum    QoS qos;char    *username;char    *password;char    *host;int     port;int     showtopics;
} opts = {(char *)"iot-dev", 0, (char *)"\n", QOS0, "admin", "password", (char *)"localhost", 1883, 0
};//初始化结构体Cloud_MQTT_t Iot_mqtt;iot_device_info_t gateway = {.iotstatus = IOT_STATUS_LOGIN,.model = {"DEVICE"},.company = {"/HB/WH/"}
};//初始化主题void iot_mqtt_init(Cloud_MQTT_t *piot_mqtt) 
{memset(piot_mqtt, '\0', sizeof(Cloud_MQTT_t));sprintf(piot_mqtt->sub_topic, "%s%s%s", gateway.model, gateway.company, MQTT_CLIENT_ID);	//将初始化好的订阅主题填到数组中printf("subscribe:%s\n", piot_mqtt->sub_topic);sprintf(piot_mqtt->pub_topic, "%s%s", "STATUS/", MQTT_CLIENT_ID);	//将初始化好的发布主题填到数组中printf("pub:%s\n", piot_mqtt->pub_topic);piot_mqtt->DataArrived_Cb = mqtt_data_rx_cb;		//设置接收到数据回调函数}
void MQTTMessageArrived_Cb(MessageData* md)
{MQTTMessage *message = md->message; Cloud_MQTT_t *piot_mqtt = &Iot_mqtt;if (NULL != piot_mqtt->DataArrived_Cb) {piot_mqtt->DataArrived_Cb((void *)message->payload, message->payloadlen);//异步消息体//设置完数据以后记得清空数据体 要不然可能会组合黏连在一起memset((void *)message->payload, 0, message->payloadlen);}
}int mqtt_device_connect(Cloud_MQTT_t *piot_mqtt)
{int rc = 0, ret = 0;NewNetwork(&piot_mqtt->Network);printf("topic = %s\n", piot_mqtt->sub_topic);rc = ConnectNetwork(&piot_mqtt->Network, MQTT_HOST, (int)MQTT_PORT);	if (rc != 0) {printf("mqtt connect network fail \n");ret = -101;goto __END;}else {printf("mqtt network Connection succeeded \n");}MQTTClient(&piot_mqtt->Client, &piot_mqtt->Network, COMMAND_TIMEOUT_MS, piot_mqtt->mqtt_buffer, MQTT_BUF_SIZE, piot_mqtt->mqtt_read_buffer, MQTT_BUF_SIZE);MQTTPacket_connectData data = MQTTPacket_connectData_initializer;if (piot_mqtt->willFlag) {data.willFlag = 1;memcpy(&data.will, &piot_mqtt->will, sizeof(MQTTPacket_willOptions));} else {data.willFlag = 0;}data.MQTTVersion = 3;data.clientID.cstring = MQTT_CLIENT_ID;//data.username.cstring = MQTT_USER;//data.password.cstring = MQTT_PASS;data.keepAliveInterval = 60;data.cleansession = 1;rc = MQTTConnect(&piot_mqtt->Client, &data);if (rc) {printf("连接MQTT_Broker[%s]失败！\n", (char *)MQTT_HOST);printf("rc = %d\n", rc);ret = -102;goto __END;}else{printf("连接MQTT_Broker[%s]成功! \n", (char *)MQTT_HOST);}rc = MQTTSubscribe(&piot_mqtt->Client, piot_mqtt->sub_topic, opts.qos, MQTTMessageArrived_Cb);if (rc) {printf("订阅服务端主题失败\n");ret = -105;goto __END;}gateway.iotstatus = IOT_STATUS_CONNECT;printf("订阅服务端主题成功，ret = %d\n", rc);__END:return ret;
}int mqtt_device_disconnect(Cloud_MQTT_t *piot_mqtt)//断开mqtt连接
{int ret = 0;ret = MQTTDisconnect(&piot_mqtt->Client);printf("已断开MQTT_Broker连接, ret = %d\n", ret);return ret;
}void iot_yield(Cloud_MQTT_t *piot_mqtt)
{int ret;switch (gateway.iotstatus) {case IOT_STATUS_LOGIN:ret = mqtt_device_connect(piot_mqtt);if (ret < 0) {printf("MQTT_Broker连接异常, error code %d\n", ret);sleep(1);}break;case IOT_STATUS_CONNECT:ret = MQTTYield(&piot_mqtt->Client, 100);if (ret != SUCCESS) {gateway.iotstatus = IOT_STATUS_DROP;}break;case IOT_STATUS_DROP:printf("连接MQTT_Broker失败\n");mqtt_device_disconnect(piot_mqtt);gateway.iotstatus = IOT_STATUS_LOGIN;usleep(1000);break;default:break;}
}int mqtt_will_msg_set(Cloud_MQTT_t *piot_mqtt, char *pbuf, int len)//设置遗嘱函数
{memset(piot_mqtt->will_topic, '\0', MQTT_TOPIC_SIZE);MQTTPacket_willOptions mqtt_will = MQTTPacket_willOptions_initializer;strcpy(piot_mqtt->will_topic, Iot_mqtt.pub_topic);memcpy(&Iot_mqtt.will, &mqtt_will, sizeof(MQTTPacket_willOptions));Iot_mqtt.willFlag = 1;Iot_mqtt.will.retained = 1;Iot_mqtt.will.topicName.cstring = (char *)piot_mqtt->will_topic;Iot_mqtt.will.message.cstring = (char *)pbuf;Iot_mqtt.will.qos = QOS2;}void mqtt_data_rx_cb(void *pbuf, int len) 
{printf("out len = %d\n", len);	//打印接收到的数据printf("out data = %s\n\n\n", (unsigned char *)pbuf);	//打印接收到的数据
}int mqtt_data_write(char *pbuf, int len, char retain)
{Cloud_MQTT_t *piot_mqtt = &Iot_mqtt; int ret = 0;MQTTMessage message;char my_topic[128] = {0};strcpy(my_topic, piot_mqtt->pub_topic);printf("publish topic is :%s\r\n", my_topic);printf("mqtt tx len = %d\r\n", len);printf("mqtt tx len = %s\r\n\n", pbuf);message.payload = (void *)pbuf;message.payloadlen = len;message.dup = 0;message.qos = QOS1;if (retain) {message.retained = 1;} else {message.retained = 0;}ret = MQTTPublish(&piot_mqtt->Client, my_topic, &message);	//发布一个主题return ret;
}void *cloud_mqtt_thread(void *arg)
{int ret, len; char will_msg[256] = {"hello"};						//初始化遗嘱数据iot_mqtt_init(&Iot_mqtt);									//初始化主题mqtt_will_msg_set(&Iot_mqtt, will_msg, strlen(will_msg));	//设置遗嘱ret = mqtt_device_connect(&Iot_mqtt);						//初始化并连接mqtt服务器while (ret < 0) {printf("ret = %d\r\n", ret);sleep(3);ret = mqtt_device_connect(&Iot_mqtt);}while (1){iot_yield(&Iot_mqtt);									//维持服务器稳定，断开重连}
}

 7.mqtt.h

#ifndef NET_PROC_H
#define NET_PROC_H#ifdef __cplusplus
extern "C" {
#endifenum  iot_ctrl_status_t
{IOT_STATUS_LOGIN,IOT_STATUS_CONNECT,IOT_STATUS_DROP,
};typedef void  (*pMessageArrived_Fun)(void*,int len);void mqtt_module_init(void);
int mqtt_data_write(char *pbuf, int len, char retain);void mqtt_data_rx_cb(void *pbuf, int len);
void *cloud_mqtt_thread(void *arg);
#define mDEBUG(fmt, ...)  printf("%s[%s](%d):" fmt,__FILE__,__FUNCTION__,__LINE__,##__VA_ARGS__)#ifdef __cplusplus
}
#endif#endif

 8.最后我们再来看一下调用:

 整个项目有几个难点需要解决:

1.pthread库的报错,详情看博客CSDN

2.curl库的导入

3.ssl库的导入

4.crypto库的导入

 因为我是在arm上跑的,所以我在我的arm主板的系统里面找到这仨库,然后通过配置使用,

别忘了头文件的配置

 我碰上的异常情况有两个:

1.其实我这个项目代码是在csdn上下载的资源,但是我发现我连不上我的mqtt服务器,后来发现,是因为使用国外的服务器,需要的超时时间更长,我就把超时时间改成了10秒;如下图1

2.一开始我把源码下载下来的时候,我没在看到mqtt.c里面有设置完数据以后记得清空数据体,所以我除了第一次启动,后面接收到的数据都有多余,后来查看发现,因为是代码里面没有清除数据体导致的!如下图2: 

 图1:

图2: