小编对Linux这一块的实际开发经验虽然还不是很足，但也知道进程间通信有那么几种方式：管道、消息队列、共享内存、套接字等。
某项目中使用了MQTT作为进程间的通信方式，感觉很新颖，网上好像也比较少有看到使用MQTT来作为进程间通信的方式的。这两篇笔记我们就来一起学习一下这种方式。
MQTT的一些介绍
❝
以下介绍内容来自《[野火]《LwIP应用开发实战指南》
❞
MQTT 协议全称是 Message Queuing Telemetry Transport，翻译过来就是消息队列遥测传输协议，它是物联网常用的应用层协议，运行在 TCP/IP 中的应用层中，依赖 TCP 协议，因此它具有非常高的可靠性，同时它是基于 TCP 协议的模型发布/订阅主题消息的轻量级协议。
1、MQTT通信模型
MQTT协议是基于客户端-服务器模型，在协议中主要有三种身份：发布者(Publisher)、服务器(Broker) 以及订阅者(Subscriber)。  并且消息发布者可以同时是订阅者 。
MQTT 消息的发布者和订阅者都是客户端，服务器只是作为一个中转的存在，将发布者发布的消息进行转发给所有订阅该主题的订阅者。
MQTT 客户端的功能：
发布消息给其它相关的客户端。
订阅主题请求接收相关的应用消息。
取消订阅主题请求移除接收应用消息。
从服务端终止连接。
MQTT 服务器常被称为 Broker(消息代理) 。它的功能有：
接受来自客户端的网络连接请求。
接受客户端发布的应用消息。
处理客户端的订阅和取消订阅请求。
转发应用消息给符合条件的已订阅客户端(包括发布者自身)。
2、MQTT消息
MQTT所发的消息包含：主题+内容，客户端可以订阅任意主题，若有其它客户端发布主题时符合所订阅的主题，就会由网关发送到客户端。
「什么是主题？」
MQTT 服务器为每个连接的客户端(订阅者)添加一个标签，该标签与服务器中的所有订阅相匹配， 服务器会将消息转发给与标签相匹配的每个客户端。这样的一个标签就是主题。
「服务质量：」
MQTT提供三种服务质量(Quality of Service，简写QoS)，供开发者根据不同的情景选择不同的服务级别：
QoS0：最多发送一次消息，在消息发送出去后，接收者不会发送回应，发送者也不会重发消息。
QoS1：最少发送一次消息(消息最少需要送达一次，也有可送达多次)， QoS 1的 PUBLISH 报文的可变报头中包含一个报文标识符，需要 PUBACK 报文确认。
QoS2：这是最高等级的服务质量，消息丢失和重复都是不可接受的。只不过使用这个服务质量等级会有额外的开销，这个等级常用于支付中，因为支付是必须有且仅有一次成功，总不能没给钱或者给了多次钱吧。
mosquitto的使用
1、mosquitto简介
mosquitto是一款开源的MQTT消息代理(服务器)软件，提供轻量级的，支持可发布/可订阅的的消息推送模式，使设备对设备之间的短消息通信变得简单，比如现在应用广泛的低功耗传感器，手机、嵌入式计算机、微型控制器等移动设备。
mosquitto仓库地址：
❝
https://gitee.com/zhengnianli/mosquitto
❞
或
❝
https://github.com/eclipse/mosquitto
❞
2、mosquitto实践
从以上链接下载mosquitto源码，得到：
在mosquitto路径下，依次输入如下命令编译：
mkdir build
cd build
cmake ../
make
若未安装cmake则需要自行安装，可参考往期文章：《面试官：Linux下如何编译C程序？》
若执行cmake ../指令时出现Could NOT find OpenSSL的问题：
可输入如下命令安装OpenSSL(Ubuntu下)：
❝
sudo apt-get install libssl-dev
❞
执行make编译完成后，我们可以看到build目录下的client与src文件夹下会生成一些可执行文件：
我们重点关注mosquitto_pub、mosquitto_sub、mosquitto这三个可执行文件。其中mosquitto是服务器软件，mosquitto_pub是发布者客户端，mosquitto_sub订阅者客户端。
下面我们来简单测试一下：
其中，mosquitto_pub对应的源文件为mosquitto/client/pub_client.c，mosquitto_sub对应的源文件为mosquitto/client/sub_client.c，感兴趣的朋友可自行阅读学习。
这个示例其实就是两个进程间的通信，前提是需要一个本地代理服务器。如果需要应用于我们嵌入式Linux中的进程间通信，需要使用交叉编译器编译出一个可运行在我们的arm板上的mosquitto服务器。
以上就是本次的分享，下一篇我们再一起看更具体的实例。如有错误，欢迎指出，谢谢！
免责声明：文章部分内容来源网络，免费传达知识，版权归原作者所有。如涉及作品版权问题，请联系我进行删除。
猜你喜欢
wireshark抓包工具的使用及分析
什么是Linux内核空间与用户空间？
1024G 嵌入式资源大放送！包括但不限于C/C++、单片机、Linux等。在公众号聊天界面回复1024，即可免费获取！

简单说一下MQTT，MQTT协议目前是流行的物联网传输协议，它也是基于发布/订阅编程模式的消息协议，很多nb-iot模块也能支持mqtt协议，那么如何使用mqtt客户端和MQTT服务器端进行交互呢？
首先在嵌入式linux系统安装python环境使用pip安装paho-mqtt
pip install paho-mqtt
或源码安装
git clone https://github.com/eclipse/paho.mqtt.python
 进入paho.mqtt.python目录执行：
python setup.py install
为了做测试mqtt客户端，需要安装一个mqtt Broker服务器,可以使用mosquitto，它是个实现了MQTT3.1协议的Broker服务器。官网下载源码：http://mosquitto.org/download
在安装之后，程序默认监听1883端口
mosquitto -v -c /etc/mosquitto/mosquitto.conf #启动
基于mqtt协议的发布/订阅测试
测试发布：
# encoding: utf-8import paho.mqtt.client as mqttHOST = "192.168.1.100"PORT = 1883def client():client = mqtt.Client()client.connect(HOST, PORT, 60)client.publish("sensor_1

本文改编于

Linux下学习用C语言实现MQTT(三）（异步函数）
https://blog.csdn.net/qq_43260665/article/details/88541433

原文是执行一次发送一次消息，然后就断开了，

我把它改成了可以连续发送消息的mqtt实现：

Talk is cheap show me the code!

MQTT      publish.cpp

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#include <unistd.h>

#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ClientPub"
#define TOPIC       "MQTT"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L

volatile MQTTAsync_token deliveredtoken;

int finished = 0;
bool is_connect = false;
void connlost(void *context, char *cause)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int rc;

	printf("\nConnection lost\n");
	printf("     cause: %s\n", cause);

	printf("Reconnecting\n");
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
 		finished = 1;
	}
}


void onDisconnect(void* context, MQTTAsync_successData* response)
{
	printf("Successful disconnection\n");
	finished = 1;
}


void onSend(void* context, MQTTAsync_successData* response)
{
	//MQTTAsync client = (MQTTAsync)context;
	//MQTTAsync_disconnectOptions opts = MQTTAsync_disconnectOptions_initializer;
	//int rc;

	printf("Message with token value %d delivery confirmed\n", response->token);
      
	//opts.onSuccess = onDisconnect;
	//opts.context = client;
        /* 
	if ((rc = MQTTAsync_disconnect(client, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start sendMessage, 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);
	finished = 1;
}


void onConnect(void* context, MQTTAsync_successData* response)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	int rc;

	printf("Successful connection\n");
	is_connect = true;
/*	opts.onSuccess = onSend;
	opts.context = client;

	pubmsg.payload = (void*)PAYLOAD;
	pubmsg.payloadlen = (int)strlen(PAYLOAD);
	pubmsg.qos = QOS;
	pubmsg.retained = 0;
	deliveredtoken = 0;
        
	    if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
	    {
		printf("Failed to start sendMessage, return code %d\n", rc);
		exit(EXIT_FAILURE);
	    }
*/        
}


int main(int argc, char* argv[])
{
	MQTTAsync client;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int rc;

	MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

	MQTTAsync_setCallbacks(client, NULL, connlost, NULL, NULL);

	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnect;
	conn_opts.onFailure = onConnectFailure;
	conn_opts.context = client;
        printf("before\n");
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}
        printf("after\n");
	printf("Waiting for publication of %s\n"
         "on topic %s for client with ClientID: %s\n",
         PAYLOAD, TOPIC, CLIENTID);
        while(!is_connect)
	{
	   printf("sleep\n");
            usleep(10000L);
	}
        
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	   
        opts.onSuccess = onSend;
	opts.context = client;

        while(1)
	{
	   
	   int rc;
           printf("while1\n");
           //#define MQTTAsync_message_initializer 
	   MQTTAsync_message pubmsg = MQTTAsync_message_initializer;
	   pubmsg.payload = (void*)PAYLOAD;
	   pubmsg.payloadlen = (int)strlen(PAYLOAD);
	   pubmsg.qos = QOS;
	   pubmsg.retained = 0;
	   deliveredtoken = 0;

	   if ((rc = MQTTAsync_sendMessage(client, TOPIC, &pubmsg, &opts)) != MQTTASYNC_SUCCESS)
	   {
		printf("Failed to start sendMessage, return code %d\n", rc);
		exit(EXIT_FAILURE);
	   }
	   usleep(100000L);
	}

	while (!finished)
		#if defined(WIN32)
			Sleep(100);
		#else
			usleep(10000L);
		#endif

	MQTTAsync_destroy(&client);
 	return rc;
}
  


 

 

subscruib.cpp

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTAsync.h"
#include <unistd.h>


#define ADDRESS     "tcp://localhost:1883"
#define CLIENTID    "ClientSub"
#define TOPIC       "MQTT"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L

volatile MQTTAsync_token deliveredtoken;

int disc_finished = 0;
int subscribed = 0;
int finished = 0;

void connlost(void *context, char *cause)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	int rc;

	printf("\nConnection lost\n");
	if (cause)
		printf("     cause: %s\n", cause);

	printf("Reconnecting\n");
	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		finished = 1;
	}
}


int msgarrvd(void *context, char *topicName, int topicLen, MQTTAsync_message *message)
{
    int i;
    char* payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = (char*)message->payload;
    for(i=0; i<message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');
    MQTTAsync_freeMessage(&message);
    MQTTAsync_free(topicName);
    return 1;
}


void onDisconnect(void* context, MQTTAsync_successData* response)
{
	printf("Successful disconnection\n");
	disc_finished = 1;
}


void onSubscribe(void* context, MQTTAsync_successData* response)
{
	printf("Subscribe succeeded\n");
	subscribed = 1;
}

void onSubscribeFailure(void* context, MQTTAsync_failureData* response)
{
	printf("Subscribe failed, rc %d\n", response ? response->code : 0);
	finished = 1;
}


void onConnectFailure(void* context, MQTTAsync_failureData* response)
{
	printf("Connect failed, rc %d\n", response ? response->code : 0);
	finished = 1;
}


void onConnect(void* context, MQTTAsync_successData* response)
{
	MQTTAsync client = (MQTTAsync)context;
	MQTTAsync_responseOptions opts = MQTTAsync_responseOptions_initializer;
	int rc;

	printf("Successful connection\n");

	printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
	opts.onSuccess = onSubscribe;
	opts.onFailure = onSubscribeFailure;
	opts.context = client;

	deliveredtoken = 0;

	if ((rc = MQTTAsync_subscribe(client, TOPIC, QOS, &opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start subscribe, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}
}


int main(int argc, char* argv[])
{
	MQTTAsync client;
	MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
	MQTTAsync_disconnectOptions disc_opts = MQTTAsync_disconnectOptions_initializer;
	int rc;
	int ch;

	MQTTAsync_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

	MQTTAsync_setCallbacks(client, client, connlost, msgarrvd, NULL);

	conn_opts.keepAliveInterval = 20;
	conn_opts.cleansession = 1;
	conn_opts.onSuccess = onConnect;
	conn_opts.onFailure = onConnectFailure;
	conn_opts.context = client;
	if ((rc = MQTTAsync_connect(client, &conn_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start connect, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}

	while	(!subscribed)
		#if defined(WIN32)
			Sleep(100);
		#else
			usleep(10000L);
		#endif

	if (finished)
		goto exit;

	do 
	{
		ch = getchar();
	} while (ch!='Q' && ch != 'q');

	disc_opts.onSuccess = onDisconnect;
	if ((rc = MQTTAsync_disconnect(client, &disc_opts)) != MQTTASYNC_SUCCESS)
	{
		printf("Failed to start disconnect, return code %d\n", rc);
		exit(EXIT_FAILURE);
	}
 	while	(!disc_finished)
		#if defined(WIN32)
			Sleep(100);
		#else
			usleep(10000L);
		#endif

exit:
	MQTTAsync_destroy(&client);
 	return rc;
}


 

一、前言：

mqtt协议是轻量级的消息订阅和发布（publish/subscribe）协议，建立在TCP/IP协议之上，在物联网应用中广泛使用。
二、源码下载：

链接：https://pan.baidu.com/s/1S1pT_ZZURg21DF5mIBg3pw 密码：tqgh
二使用说明：

1.下载解压出来进入mqtt文件夹内容如图src.png所示：



2.将mqtt文件夹拷贝linux ununtu下面：

在ununtu终端下根据如下步骤执行命令：

1)make clean //清理项目

2)vim mqtt.c //打开mqtt.c文件将如图：ip.png所示格式修改为自己的服务器ip地址、端口、用户名和密码：



ip.png
wq保存退出；

3)make //编译项目

./mqtt_demo//运行成功如图data.png所示(运行之前需要运行windows paho帮助测试，在本文下面有介绍)



data.png
①：订阅主题：2017/my/todev

②：发布主题：2017/my/toapp

③：接收到数据打印:asdfafs

④：ctrl + c//结束运行
3.运行windows paho来测试mqtt_demo(没有安装可以安装一下连接进行安装：https://www.jianshu.com/p/48c36b72fec2)：

1）首先根据图:login.png所示：点击左上角绿色+号->点击选项->输入用户名和密码->点击MQTT返回主主界面



login.png
2）图msg.png步骤解析如下：

①：输入服务器ip地址和端口号

②：点击连接

③：点击绿色小+号订阅主题

④：输入主题名字

⑤：把需要订阅的主题勾上

⑥：点击订阅

⑦：输入发布主题

⑧：输入发布的内容

⑨：点击发布

⑩：接收到订阅该主题发布出来的内容(linux 下面客户端程序发出来的)

⑪:自己发布出去的内容

⑫:linux 客户端掉线发布出来的遗嘱