【MQTT协议使用总结】基于-FreeRTOS平台-移植MQTT协议栈

仓库地址

https://github.com/eclipse/paho.mqtt.embedded-c

这里官方给了一些平台适配案例，这里参考FreeRTOS的

关键接口适配

使用使用了lwip的一些接口

FreeRTOS_read

int FreeRTOS_read(Network* n, unsigned char* buffer, int len, int timeout_ms)
{// 将超时时间转换为 timeval 结构体表示的时间间隔struct timeval interval = {timeout_ms / 1000, (timeout_ms % 1000) * 1000};// 如果超时时间小于等于 0，将超时时间设置为 0.1 秒（100 微秒）if (interval.tv_sec < 0 || (interval.tv_sec == 0 && interval.tv_usec <= 0)) {interval.tv_sec = 0;interval.tv_usec = 100;}fd_set read_fds;// 初始化文件描述符集合，清空 read_fdsFD_ZERO(&read_fds);// 将网络结构体中的套接字加入到文件描述符集合中，表示关注该套接字的可读事件FD_SET(n->my_socket, &read_fds);int bytes = 0;while(bytes < len) {// 使用 lwip_select 等待套接字可读事件，设置超时时间为 intervalint result = lwip_select(n->my_socket + 1, &read_fds, NULL, NULL, &interval);// 如果返回值小于等于 0，表示没有可读事件发生或者发生错误，跳出循环if (result <= 0)break;// 套接字可读int rc = lwip_recv(n->my_socket, &buffer[bytes], (len - bytes), 0);// 如果接收返回值为 -1，表示接收出错if (rc == -1) {// 如果错误不是 EAGAIN（资源暂时不可用）或 EWOULDBLOCK（操作会阻塞）if (errno!= EAGAIN && errno!= EWOULDBLOCK)bytes = -1;break;}// 如果接收返回值为 0，表示连接已关闭else if (rc == 0) {bytes = 0;break;}// 如果接收成功，更新已接收的字节数elsebytes += rc;}return bytes;
}

FreeRTOS_write

int FreeRTOS_write(Network* n, unsigned char* buffer, int len, int timeout_ms)
{TickType_t xTicksToWait = timeout_ms / portTICK_PERIOD_MS; /* convert milliseconds to ticks */TimeOut_t xTimeOut;int sentLen = 0;int flags = 0;vTaskSetTimeOutState(&xTimeOut); /* Record the time at which this function was entered. */do {int rc = 0;// 设置 lwip_send 为非阻塞方式flags = MSG_DONTWAIT;rc = lwip_send(n->my_socket, buffer + sentLen, len - sentLen, flags);if (rc > 0)sentLen += rc;else if (rc == 0 || (rc < 0 && errno == EAGAIN)) {// 缓冲区已满，等待一段时间再尝试发送vTaskDelay(pdMS_TO_TICKS(10)); // 延时10ms再尝试发送} else if (rc < 0) {sentLen = rc;break;}} while (sentLen < len && xTaskCheckForTimeOut(&xTimeOut, &xTicksToWait) == pdFALSE);return sentLen;
}

NetworkInit && NetworkConnect && NetworkDisconnect

// 初始化网络结构体
void NetworkInit(Network* n)
{// 将套接字初始化为无效值n->my_socket = -1;// 设置网络结构体的读取函数指针为 FreeRTOS_readn->mqttread = FreeRTOS_read;// 设置网络结构体的写入函数指针为 FreeRTOS_writen->mqttwrite = FreeRTOS_write;
}// 连接到指定地址和端口
int NetworkConnect(Network* n, char* addr, int port)
{// 打开连接并获取套接字描述符n->my_socket = transport_open((int8_t*)addr, port);// 如果套接字描述符有效（不等于 -1），表示连接成功if (-1!= n->my_socket) {return 0;}// 连接失败，返回 -1return -1;
}// 断开网络连接
void NetworkDisconnect(Network* n)
{// 如果套接字描述符有效（大于 0）if (n->my_socket > 0 ) {// 关闭套接字closesocket(n->my_socket);// 将套接字描述符重置为无效值n->my_socket = -1;}
}

int32_t transport_open(int8_t* servip, int32_t port)
{int32_t *sock = &mysock;int32_t ret;int32_t opt;struct sockaddr_in addr;// 初始化服务器信息memset(&addr, 0, sizeof(addr));addr.sin_len = sizeof(addr);addr.sin_family = AF_INET;// 填写服务器端口号addr.sin_port = PP_HTONS(port);// 填写服务器 IP 地址addr.sin_addr.s_addr = inet_addr((const char*)servip);// 创建套接字*sock = socket(AF_INET, SOCK_STREAM, 0);// 连接服务器ret = connect(*sock, (struct sockaddr*)&addr, sizeof(addr));if(ret!= 0){// 连接失败，关闭链接close(*sock);// 返回 -1 表示连接失败return -1;}// 连接成功，设置 TCP_NODELAY 选项（禁用 Nagle 算法）opt = 1;setsockopt(*sock, IPPROTO_TCP, TCP_NODELAY, (void *)&opt, sizeof(int));// 返回套接字描述符return *sock;
}

总结

实际使用过程中，遇到了lwip阻塞的阻塞情况，注意采用非阻塞方式进行读取。