博客主页https://blog.csdn.net/wkd_007博客内容嵌入式开发、Linux、C语言、C、数据结构、音视频⏰发布时间⏰2024-05-13 12:43:09本文未经允许不得转发目录一、概述二、paho.mqtt.c库介绍✨2.1 异步的paho.mqtt.c库(Asynchronous MQTT client library for C)✨2.2 同步的paho.mqtt.c库(MQTT Client library for C)✨2.3 异步和同步库的区别(Asynchronous vs synchronous client applications)三、paho.mqtt.c 库的下载、交叉编译✨3.1 openssl 的下载、交叉编译✨3.2 paho.mqtt.c 交叉编译四、paho.mqtt.c库写一个MQTT客户端✨4.1 订阅——MQTTAsync_subscribe.c✨4.2 发布——MQTTAsync_publish.c五、总结一、概述上篇文章 【MQTT】mosquitto 的 “下载、交叉编译、使用” 详细教程手把手搭建一个MQTT Broker 介绍了怎样搭建一个MQTT Broker(代理)。MQTT Broker(代理)的搭建很重要一般很少自己开发一个MQTT Broker而是搭建好一个开源的MQTT Broker(代理)然后其他的设备或机器都作为MQTT客户端让MQTT Broker来转发消息。关于MQTT协议的作为编程人员更多的是开发MQTT的客户端本文介绍的paho.mqtt.c就是开发MQTT客户端常用的开源库之一。paho.mqtt.c是Eclipse Paho项目的一个开源库。Eclipse Paho项目提供了以各种编程语言实现的MQTT和MQTT-SN的开源库主要是客户端。下面是这个项目的MQTT客户端开源库想要了解更多Eclipse Paho项目的资料可以到其官网https://eclipse.dev/paho/二、paho.mqtt.c库介绍paho.mqtt.c库编译后会生成四个动态库libpaho-mqtt3a.so异步的paho.mqtt.c库libpaho-mqtt3as.so异步的、使用了SSL的paho.mqtt.c库libpaho-mqtt3c.so经典的、同步的paho.mqtt.c库libpaho-mqtt3cs.so经典的、同步的、使用了SSL的paho.mqtt.c库✨2.1 异步的paho.mqtt.c库(Asynchronous MQTT client library for C)关于异步的paho.mqtt.c库可以参考其官网文章Asynchronous MQTT client library for C这里只摘要了使用方法异步的paho.mqtt.c库有如下相似的使用框架1.创建一个客户端对象2.设置连接MQTT服务器的选项3.设置回调函数4.将客户端和服务器连接5.订阅客户端需要接收的所有话题6.重复以下操作直到结束a.发布客户端需要的任意信息b.处理所有接收到的信息7.断开客户端连接8.释放客户端使用的所有内存。✨2.2 同步的paho.mqtt.c库(MQTT Client library for C)关于异步的paho.mqtt.c库可以参考其官网文章MQTT Client library for C这里只摘要了使用方法同步的paho.mqtt.c库有如下相似的使用框架1.创建一个客户端对象2.设置连接MQTT服务器的选项3.如果要使用多线程(异步模式)操作则调用 MQTTClient_setCallbacks() 设置回调函数4.订阅客户端需要接收的任意话题5.重复以下操作直到结束a.发布客户端需要的任意信息b.处理所有接收到的信息6.断开客户端连接7.释放客户端使用的所有内存。✨2.3 异步和同步库的区别(Asynchronous vs synchronous client applications)下面是译文原文地址Asynchronous vs synchronous client applications客户端库支持两种操作模式。这些模式被称为同步模式和异步模式。如果您的应用程序调用 MQTTClient_setCallbacks()这将使客户端进入异步模式否则它将以同步模式运行。在同步模式下客户端应用程序在单个线程上运行。消息是使用 MQTTClient_publish() 和 MQTTClient_publishMessage() 函数发布的。要确定QoS1或QoS2请参阅Quality of service消息已成功传递应用程序必须调用 MQTTClient_waitForCompletion() 函数。Synchronous publication example 中展示了一个同步发布的示例。在同步模式下接收消息使用MQTTClient_receive()函数。客户端应用程序必须相对频繁地调用MQTTClient_ereceived()或MQTTClient_yield()以便允许执行确认和保持与服务器的网络连接的 MQTT “ping” 操作。在异步模式下客户端应用程序在多个线程上运行。主程序像同步模式一样地调用客户端库中的函数来发布和订阅。然而握手和保持网络连接的处理是在后台执行的。通过调用 MQTTClient_setCallbacks()使用在库中注册的回调将状态和消息接收通知提供给客户端应用程序请参阅 MQTTClient_messageArrived()、MQTTClient_connectionLost() 和MQTTClient_edeliveryComplete() 。然而这个API不是线程安全的——在没有同步的情况下不可能从多个线程调用它。您可以使用MQTTAsync API来实现这一点。三、paho.mqtt.c 库的下载、交叉编译本文下载的是paho.mqtt.c-1.3.13.tar.gz下载地址https://github.com/eclipse/paho.mqtt.c/archive/refs/tags/v1.3.13.tar.gz编译 paho.mqtt.c 库之前需要先编译其依赖库openssl。✨3.1 openssl 的下载、交叉编译本文下载的是openssl-OpenSSL_1_1_1g.tar.gz下载地址https://codeload.github.com/openssl/openssl/tar.gz/refs/tags/OpenSSL_1_1_1g为什么使用这么旧的版本因为这个我之前编译过而且使用没问题。编译步骤1、解压缩tarzxf openssl-OpenSSL_1_1_1g.tar.gz2、进入目录并配置输出目录和交叉编译器, (linux-generic32表示是32位操作系统,个别文章加了这个选项就不用去掉 -m64我这里行不通)cdopenssl-OpenSSL_1_1_1g/ ./config no-asm shared no-async--prefixpwd/ssl_result --cross-compile-prefixaarch64-mix210-linux-3、执行下面命令删除Makefile文件的-m64sed-is/-m64//Makefile执行后可以避免出现这个编译错误aarch64-mix210-linux-gcc: error: unrecognized command line option -m644、编译、安装makemakeinstall成功编译后在openssl-OpenSSL_1_1_1g/目录会生成一个ssl_result目录可以看到里面生成的库✨3.2 paho.mqtt.c 交叉编译编译步骤1、解压缩创建要安装目录paho.mqtt.c_resulttarzxf paho.mqtt.c-1.3.13.tar.gzmkdirpaho.mqtt.c_result/bin-pmkdirpaho.mqtt.c_result/include-pmkdirpaho.mqtt.c_result/lib-pmkdirpaho.mqtt.c_result/share/man/man1-p2、进入目录交叉编译cdpaho.mqtt.c-1.3.13/makeCCaarch64-mix210-linux-gcc CFLAGS:-Ipwd/../ssl_result/includeLDFLAGS:-Lpwd/../ssl_result/libCFLAGS:“-I pwd/…/ssl_result/include”指定前面编译的 openssl 的头文件LDFLAGS:“-L pwd/…/ssl_result/lib”指定前面编译的 openssl 的库文件路径3、make install安装编译结果makeinstallprefixpwd/../paho.mqtt.c_resultprefixpwd/…/paho.mqtt.c_result指定安装目录路径编译完成后会生成 目录内容如下四、paho.mqtt.c库写一个MQTT客户端下面提供两个使用了 paho.mqtt.c库(libpaho-mqtt3a) MQTT客户端的例子。运行下面代码前要先搭建一个MQTT Broker参考上篇文章【MQTT】mosquitto 的 “下载、交叉编译、使用” 详细教程手把手搭建一个MQTT Broker✨4.1 订阅——MQTTAsync_subscribe.c这是使用了 libpaho-mqtt3a.so 进行订阅消息的源码源码路径在源码的这个路径paho.mqtt.c-1.3.13/src/samples/MQTTAsync_subscribe.c只更改了服务器地址。完整代码如下/******************************************************************************* * Copyright (c) 2012, 2022 IBM Corp., Ian Craggs * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 * and Eclipse Distribution License v1.0 which accompany this distribution. * * The Eclipse Public License is available at * https://www.eclipse.org/legal/epl-2.0/ * and the Eclipse Distribution License is available at * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Ian Craggs - initial contribution *******************************************************************************/#includestdio.h#includestdlib.h#includestring.h#includeMQTTAsync.h#if!defined(_WIN32)#includeunistd.h#else#includewindows.h#endif#ifdefined(_WRS_KERNEL)#includeOsWrapper.h#endif#defineADDRESS192.168.3.227:1883#defineCLIENTIDExampleClientSub#defineTOPICMQTT Examples#definePAYLOADHello World!#defineQOS1#defineTIMEOUT10000Lintdisc_finished0;intsubscribed0;intfinished0;voidonConnect(void*context,MQTTAsync_successData*response);voidonConnectFailure(void*context,MQTTAsync_failureData*response);voidconnlost(void*context,char*cause){MQTTAsync client(MQTTAsync)context;MQTTAsync_connectOptions conn_optsMQTTAsync_connectOptions_initializer;intrc;printf(\nConnection lost\n);if(cause)printf( cause: %s\n,cause);printf(Reconnecting\n);conn_opts.keepAliveInterval20;conn_opts.cleansession1;conn_opts.onSuccessonConnect;conn_opts.onFailureonConnectFailure;if((rcMQTTAsync_connect(client,conn_opts))!MQTTASYNC_SUCCESS){printf(Failed to start connect, return code %d\n,rc);finished1;}}intmsgarrvd(void*context,char*topicName,inttopicLen,MQTTAsync_message*message){printf(Message arrived\n);printf( topic: %s\n,topicName);printf( message: %.*s\n,message-payloadlen,(char*)message-payload);MQTTAsync_freeMessage(message);MQTTAsync_free(topicName);return1;}voidonDisconnectFailure(void*context,MQTTAsync_failureData*response){printf(Disconnect failed, rc %d\n,response-code);disc_finished1;}voidonDisconnect(void*context,MQTTAsync_successData*response){printf(Successful disconnection\n);disc_finished1;}voidonSubscribe(void*context,MQTTAsync_successData*response){printf(Subscribe succeeded\n);subscribed1;}voidonSubscribeFailure(void*context,MQTTAsync_failureData*response){printf(Subscribe failed, rc %d\n,response-code);finished1;}voidonConnectFailure(void*context,MQTTAsync_failureData*response){printf(Connect failed, rc %d\n,response-code);finished1;}voidonConnect(void*context,MQTTAsync_successData*response){MQTTAsync client(MQTTAsync)context;MQTTAsync_responseOptions optsMQTTAsync_responseOptions_initializer;intrc;printf(Successful connection\n);printf(Subscribing to topic %s\nfor client %s using QoS%d\n\nPress QEnter to quit\n\n,TOPIC,CLIENTID,QOS);opts.onSuccessonSubscribe;opts.onFailureonSubscribeFailure;opts.contextclient;if((rcMQTTAsync_subscribe(client,TOPIC,QOS,opts))!MQTTASYNC_SUCCESS){printf(Failed to start subscribe, return code %d\n,rc);finished1;}}intmain(intargc,char*argv[]){MQTTAsync client;MQTTAsync_connectOptions conn_optsMQTTAsync_connectOptions_initializer;MQTTAsync_disconnectOptions disc_optsMQTTAsync_disconnectOptions_initializer;intrc;intch;if((rcMQTTAsync_create(client,ADDRESS,CLIENTID,MQTTCLIENT_PERSISTENCE_NONE,NULL))!MQTTASYNC_SUCCESS){printf(Failed to create client, return code %d\n,rc);rcEXIT_FAILURE;gotoexit;}if((rcMQTTAsync_setCallbacks(client,client,connlost,msgarrvd,NULL))!MQTTASYNC_SUCCESS){printf(Failed to set callbacks, return code %d\n,rc);rcEXIT_FAILURE;gotodestroy_exit;}conn_opts.keepAliveInterval20;conn_opts.cleansession1;conn_opts.onSuccessonConnect;conn_opts.onFailureonConnectFailure;conn_opts.contextclient;if((rcMQTTAsync_connect(client,conn_opts))!MQTTASYNC_SUCCESS){printf(Failed to start connect, return code %d\n,rc);rcEXIT_FAILURE;gotodestroy_exit;}while(!subscribed!finished)#ifdefined(_WIN32)Sleep(100);#elseusleep(10000L);#endifif(finished)gotoexit;do{chgetchar();}while(ch!Qch!q);disc_opts.onSuccessonDisconnect;disc_opts.onFailureonDisconnectFailure;if((rcMQTTAsync_disconnect(client,disc_opts))!MQTTASYNC_SUCCESS){printf(Failed to start disconnect, return code %d\n,rc);rcEXIT_FAILURE;gotodestroy_exit;}while(!disc_finished){#ifdefined(_WIN32)Sleep(100);#elseusleep(10000L);#endif}destroy_exit:MQTTAsync_destroy(client);exit:returnrc;}交叉编译aarch64-mix210-linux-gcc MQTTAsync_subscribe.c-Ipaho.mqtt.c_result/include/-Lpaho.mqtt.c_result/lib/-lpaho-mqtt3a-oMQTTAsync_subscribe在嵌入式板子运行✨4.2 发布——MQTTAsync_publish.c这是使用了 libpaho-mqtt3a.so 进行发布消息的源码源码路径在源码的这个路径paho.mqtt.c-1.3.13/src/samples/MQTTAsync_publish.c只更改了服务器地址。完整代码如下/******************************************************************************* * Copyright(c)2012,2023IBM Corp., Ian Craggs * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 * and Eclipse Distribution License v1.0whichaccompany this distribution. * * The Eclipse Public License is available at * https://www.eclipse.org/legal/epl-2.0/ * and the Eclipse Distribution License is available at * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Ian Craggs - initial contribution *******************************************************************************/#include stdio.h#include stdlib.h#include string.h#include MQTTAsync.h#if !defined(_WIN32)#include unistd.h#else#include windows.h#endif#if defined(_WRS_KERNEL)#include OsWrapper.h#endif#define ADDRESS 192.168.3.227:1883#define CLIENTID ExampleClientPub#define TOPIC MQTT Examples#define PAYLOAD Hello World!#define QOS 2#define TIMEOUT 10000Lint finished0;void connlost(void *context, char *cause){MQTTAsync client(MQTTAsync)context;MQTTAsync_connectOptions conn_optsMQTTAsync_connectOptions_initializer;int rc;printf(\nConnection lost\n);if(cause)printf( cause: %s\n, cause);printf(Reconnecting\n);conn_opts.keepAliveInterval20;conn_opts.cleansession1;if((rcMQTTAsync_connect(client,conn_opts))!MQTTASYNC_SUCCESS){printf(Failed to start connect, return code %d\n, rc);finished1;}}void onDisconnectFailure(void* context, MQTTAsync_failureData* response){printf(Disconnect failed\n);finished1;}void onDisconnect(void* context, MQTTAsync_successData* response){printf(Successful disconnection\n);finished1;}void onSendFailure(void* context, MQTTAsync_failureData* response){MQTTAsync client(MQTTAsync)context;MQTTAsync_disconnectOptions optsMQTTAsync_disconnectOptions_initializer;int rc;printf(Message send failed token %d error code %d\n, response-token, response-code);opts.onSuccessonDisconnect;opts.onFailureonDisconnectFailure;opts.contextclient;if((rcMQTTAsync_disconnect(client,opts))!MQTTASYNC_SUCCESS){printf(Failed to start disconnect, return code %d\n, rc);exit(EXIT_FAILURE);}}void onSend(void* context, MQTTAsync_successData* response){MQTTAsync client(MQTTAsync)context;MQTTAsync_disconnectOptions optsMQTTAsync_disconnectOptions_initializer;int rc;printf(Message with token value %d delivery confirmed\n, response-token);opts.onSuccessonDisconnect;opts.onFailureonDisconnectFailure;opts.contextclient;if((rcMQTTAsync_disconnect(client,opts))!MQTTASYNC_SUCCESS){printf(Failed to start disconnect, return code %d\n, rc);exit(EXIT_FAILURE);}}void onConnectFailure(void* context, MQTTAsync_failureData* response){printf(Connect failed, rc %d\n, response ? response-code:0);finished1;}void onConnect(void* context, MQTTAsync_successData* response){MQTTAsync client(MQTTAsync)context;MQTTAsync_responseOptions optsMQTTAsync_responseOptions_initializer;MQTTAsync_message pubmsgMQTTAsync_message_initializer;int rc;printf(Successful connection\n);opts.onSuccessonSend;opts.onFailureonSendFailure;opts.contextclient;pubmsg.payloadPAYLOAD;pubmsg.payloadlen(int)strlen(PAYLOAD);pubmsg.qosQOS;pubmsg.retained0;if((rcMQTTAsync_sendMessage(client,TOPIC,pubmsg,opts))!MQTTASYNC_SUCCESS){printf(Failed to start sendMessage, return code %d\n, rc);exit(EXIT_FAILURE);}}int messageArrived(void* context, char* topicName, int topicLen, MQTTAsync_message* m){/* not expecting any messages */return1;}int main(int argc, char* argv[]){MQTTAsync client;MQTTAsync_connectOptions conn_optsMQTTAsync_connectOptions_initializer;int rc;if((rcMQTTAsync_create(client,ADDRESS,CLIENTID,MQTTCLIENT_PERSISTENCE_NONE,NULL))!MQTTASYNC_SUCCESS){printf(Failed to create client object, return code %d\n, rc);exit(EXIT_FAILURE);}if((rcMQTTAsync_setCallbacks(client,client,connlost,messageArrived,NULL))!MQTTASYNC_SUCCESS){printf(Failed to set callback, return code %d\n, rc);exit(EXIT_FAILURE);}conn_opts.keepAliveInterval20;conn_opts.cleansession1;conn_opts.onSuccessonConnect;conn_opts.onFailureonConnectFailure;conn_opts.contextclient;if((rcMQTTAsync_connect(client,conn_opts))!MQTTASYNC_SUCCESS){printf(Failed to start connect, return code %d\n, rc);exit(EXIT_FAILURE);}printf(Waiting for publication of %s\non topic %s for client with ClientID: %s\n, PAYLOAD, TOPIC, CLIENTID);while(!finished)#if defined(_WIN32)Sleep(100);#elseusleep(10000L);#endifMQTTAsync_destroy(client);returnrc;}交叉编译aarch64-mix210-linux-gcc MQTTAsync_publish.c-Ipaho.mqtt.c_result/include/-Lpaho.mqtt.c_result/lib/-lpaho-mqtt3a-oMQTTAsync_publish在嵌入式板子的运行结果五、总结本文详细介绍了 paho.mqtt.c 库的“介绍、下载、交叉编译” 以及怎么使用 aho.mqtt.c 库 编写MQTT客户端例子源码。如果文章有帮助的话点赞、收藏⭐支持一波谢谢