Cleofwine | 开发者

后端技术_MQTT的高效使用场景

最近更新：2024-11-11 | 字数总计：2.8k | 阅读估时：11分钟 | 阅读量：次

MQTT代理结构
Qos
MQTT编程
1. 客户端（Pub+Sub）
MQTT总结

MQTT代理结构

Qos

服务质量（QoS）级别是消息发送方和消息接收方之间的协议，它定义了特定消息的交付保证。MQTT中有三个QoS级别：QoS 0（最多一次）；QoS 1（至少一次）；QoS 2（恰好一次）；
MQTT中讨论QoS时，需要考虑消息传递的两个方面：
1. 消息从发布客户端传递到代理；
2. 从代理到订阅客户端的消息传递；

QoS 0

最小QoS级别为零0。这个服务级别保证了一次投递。没有交付成功的保证。接收方不确认收到消息，发送方也不存储和重新传输消息。
发布者在publish完，立马就delete掉消息，不管接收端是否会收到数据

QoS 1

QoS级别1保证消息至少被接收者收到一次。发送方存储消息，直到从接收方获得确认收到消息的发布确认(PUBACK)包为止。一条消息可能被多次发送或传递。
从上图可知，发布者到服务器有一次缓存+检测ack的过程；服务器到订阅者之间还有一次缓存+检测ack的过程；
为什么QoS 1又被称为至少一次？因为假如发布者发送publish后，它是有一个定时检测ack的定时器机制在的，假如因为网络问题迟迟没有收到ack那么会重复发送一次publish，这就导致了订阅者可能会收到不止一次的消息，所以是至少一次。

QoS 2

QoS 2是MQTT中最高级别的服务。此级别保证每个消息仅被预期的收件者接收一次。QoS 2是服务级别中最安全、最慢的质量。QoS2 不仅要确保 Receiver 能收到 Sender 发送的消息，还要保证消息不重复。它的重传和应答机制就要复杂一些，同时开销也是最大的。
这种保证由发送方和接收方之间至少两个请求/响应流(四部分握手)提供。发送方和接收方使用原始发布(PUBLISH)消息的包标识符来协调消息的传递。
只有需要非常精确的消息控制的时候，会使用这种QoS2模式，比如说扣费系统。

MQTT编程

客户端（Pub+Sub）

Pub

#include <jsoncpp/json/json.h>
#define HOST "localhost"
#define PORT  1883
#define KEEP_ALIVE 60
#define QOS 1
#define USERNAME "0voice_mqtt"
#define PASSWORD "123456"
#define TOPIC_NAME "mtopic"
#define CLIENT_ID "king123456"
static bool clean_session = false;

static float get_random_float(float min, float max) {
    float random = ((float) rand()) / (float) RAND_MAX;
    return min + (random * (max - min));
}
void my_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
{
    /* Pring all log messages regardless of level. */
    printf("log_callback-> %s\n", str);
}
int main()
{
    int ret = 0;
    struct mosquitto *mosq = NULL;
    // 使用mosquitto库函数前，要先初始化，使用之后就要清除。清除函数；int mosquitto_lib_cleanup()
    mosquitto_lib_init();
    // 创建一个新的mosquitto客户端实例，新建客户端。
    // id ：用作客户端ID的字符串。如果为NULL，将生成一个随机客户端ID。如果id为NULL，clean_session必须为true。
    // clean_session：设置为true以指示代理在断开连接时清除所有消息和订阅，设置为false以指示其保留它们，
    //                客户端将永远不会在断开连接时丢弃自己的传出消息。调用mosquitto_connect或mosquitto_reconnect
    //                将导致重新发送消息。使mosquitto_reinitialise将客户端重置为其原始状态。如果id参数为NULL，则必须将其设置为true。
    //                简言之：就是断开后是否保留订阅信息true/false
    // obj： 用户指针，将作为参数传递给指定的任何回调，（回调参数）
    mosq = mosquitto_new(CLIENT_ID, clean_session, NULL);
    if(!mosq){
        printf("create client failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    //设置密码
    if(mosquitto_username_pw_set(mosq, USERNAME, PASSWORD) != MOSQ_ERR_SUCCESS) {
        printf("client username_pw_set failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }
    // 设置日志回调函数，需要时可使用
    // PUBLISH (d0, q0, r0, m0, 'topic_pose ', ... (33 bytes))
    // 可以自定义日志回调函数
    mosquitto_log_callback_set(mosq, my_log_callback);

    // 连接到MQTT代理/服务器（主题订阅要在连接服务器之后进行）
    // mosq： 有效的mosquitto实例，mosquitto_new（）返回的mosq.
    // host: 服务器ip地址
    // port：服务器的端口号
    // keepalive：保持连接的时间间隔， 单位秒。如果在这段时间内没有其他消息交换，则代理应该将PING消息发送到客户端的秒数。
    if(mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE) != MOSQ_ERR_SUCCESS){
        fprintf(stderr, "Unable to connect.\n");
         mosquitto_lib_cleanup();
        return 1;
    }
    // 网络事件循环处理函数，通过创建新的线程不断调用mosquitto_loop() 函数处理网络事件，不阻塞
    int loop = mosquitto_loop_start(mosq);
    if(loop != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto loop error\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    while(1)
    {
        /*发布消息  坐标位置、偏航信息 */
        Json::Value root;
        root["x"] = get_random_float(0, 90);
        root["y"] = get_random_float(0, 90);
        root["yaw"] = get_random_float(0, 30);
        Json::StreamWriterBuilder writerBuilder;
        std::ostringstream os;
        std::unique_ptr<Json::StreamWriter> json_writer(writerBuilder.newStreamWriter());
        json_writer->write(root, &os);
        std::string str_json = os.str();
        // 主题发布函数
        // mosq：有效的mosquitto实例，客户端
        // mid：指向int的指针。如果不为NULL，则函数会将其设置为该特定消息的消息ID。然后可以将其与发布回调一起使用，
        //      以确定何时发送消息。请注意，尽管MQTT协议不对QoS = 0的消息使用消息ID，但libmosquitto为其分配了消息ID，
        //      以便可以使用此参数对其进行跟踪。
        // topic：要发布的主题，以null结尾的字符串
        // payloadlen：有效负载的大小（字节），有效值在0到268，435，455之间；主题消息的内容长度
        // payload： 主题消息的内容，指向要发送的数据的指针，如果payloadlen >0，则它必须时有效的存储位置。
        // qos：整数值0、1、2指示要用于消息的服务质量。
        // retain：设置为true以保留消息, 可以修改该值做测试, 比如先pub然后停止，然后调用sub看看是否可以收到消息
        ret = mosquitto_publish(mosq, NULL, TOPIC_NAME, str_json.size() + 1, str_json.c_str(),
             QOS, 1);
        if(ret != MOSQ_ERR_SUCCESS) {
            printf("mosquitto_publish failed\n");
        } else {
            printf("pub %s\n", str_json.c_str());
        }
        sleep(1);
    }
    // 释放客户端
    mosquitto_destroy(mosq);
    // 使用完mosquitto函数之后，要做清除工作
    mosquitto_lib_cleanup();

    return 0;
}

Sub

#define HOST "localhost"
#define PORT  1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE  512
#define TOPIC_NUM 3

#define QOS   1
#define CLIENT_ID "king654321"   // 发布端和订阅端我们采用不同的id测试
#define USERNAME "0voice_mqtt"
#define PASSWORD "123456"
#define TOPIC_NAME "mtopic"


static bool clean_session = false;

void print_mqtt_message(const struct mosquitto_message *message)
{
    // struct mosquitto_message{
    //     int mid;
    //     char *topic;
    //     void *payload;
    //     int payloadlen;
    //     int qos;
    //     bool retain;
    // };
    if(message) {
        printf("topic:%s mid:%d, qos:%d, retain:%d payload:%s \n", \
            message->topic, message->mid, message->qos, message->retain, (char *)message->payload);
    }
}

void my_message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message)
{
    print_mqtt_message(message);
    if(message->payloadlen){
        //parse json here
        char *data=(char *)message->payload;
        cJSON *pose=cJSON_Parse(data);
        cJSON *item_x = cJSON_GetObjectItem(pose,"x"); //获取这个对象成员
        double pose_x=item_x->valuedouble;

        cJSON *item_y = cJSON_GetObjectItem(pose,"y"); //获取这个对象成员
        double pose_y = item_y->valuedouble;

        cJSON *item_yaw = cJSON_GetObjectItem(pose,"yaw"); //获取这个对象成员
        double pose_yaw = item_yaw->valuedouble;
        printf("pose_x :%lf, pose_y :%lf, pose_yaw :%lf\n\n", pose_x, pose_y, pose_yaw);

    }else{
        printf("%s (null)\n", message->topic);
    }
    fflush(stdout);
}

void my_connect_callback(struct mosquitto *mosq, void *userdata, int result)
{
    if(!result){
        printf("Connect ok, start sub %s, qos:%d\n", TOPIC_NAME, QOS);
        // 订阅主题函数
        // ①mosq：有效的mosquitto实例，客户端
        // ②mid： 指向int的指针。如果不为NULL，则函数会将其设置为该特定消息的消息ID。
        //        然后可以将其与订阅回调一起使用，以确定何时发送消；主题的消息ID
        // ③sub： 主题名称，订阅模式。
        // ④qos : 此订阅请求的服务质量。
        mosquitto_subscribe(mosq, NULL, TOPIC_NAME, QOS);
    }else{
        printf( "Connect failed, result:%d\n", result);
    }
}

void my_subscribe_callback(struct mosquitto *mosq, void *userdata, int mid, int qos_count, const int *granted_qos)
{
    int i;
    printf("Subscribed (mid: %d): %d", mid, granted_qos[0]);
    for(i = 1; i < qos_count; i++){
        printf(", %d", granted_qos[i]);
    }
    printf("\n");
}
/*
og_callback-> Client null received PUBLISH (d0, q2, r0, m6, 'topic_pose', ... (86 bytes))
log_callback-> Client null sending PUBREC (m6, rc0)
log_callback-> Client null received PUBREL (Mid: 6)
log_callback-> Client null sending PUBCOMP (m6)
PUBREC: 是MQTT（Message Queuing Telemetry Transport）协议中的一个消息类型。
    在MQTT中，当客户端发布(PUBLISH)一条消息到服务器时，服务器会给予一个确认回复(PUBACK)，
    而当客户端收到PUBACK后，会发送一个确认接收的回复(PUBREC)给服务器。这个过程用于确保消息的可靠传递。

PUBREL: 是MQTT（Message Queuing Telemetry Transport）协议中的一种消息类型。
    它是用于在发布者和订阅者之间进行可靠消息传输的确认机制之一。
    当一个客户端发送PUBREC消息作为对收到的QoS 2级别的PUBLISH消息的确认时，
    服务器将会响应一个PUBREL消息。这个PUBREL消息表示服务器已经成功接收到客户端的确认，并且可以释放相关资源。
    通过使用PUBREL消息，MQTT确保了在QoS 2级别下消息的可靠性和顺序性。

PUBCOMP: 是MQTT协议中的一种消息类型，用于确认收到的QoS 2级别的发布消息。当客户端收到QoS 2级别的发布消息后，
    会发送PUBCOMP消息作为确认，以通知服务器已成功接收和处理该消息。
*/
void my_log_callback(struct mosquitto *mosq, void *userdata, int level, const char *str)
{
    /* Pring all log messages regardless of level. */
    printf("log_callback-> %s\n", str);
}

int main()
{
    struct mosquitto *mosq = NULL;
    //libmosquitto 库初始化
    mosquitto_lib_init();
    //创建mosquitto客户端
    mosq = mosquitto_new(CLIENT_ID, clean_session, NULL);
    if(!mosq){
        printf("create client failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }
    // 设置日志回调函数，需要时可使用
    // PUBLISH (d0, q0, r0, m0, 'topic_pose ', ... (33 bytes))
    // 可以自定义日志回调函数
    mosquitto_log_callback_set(mosq, my_log_callback);

    // 连接确认回调函数，当代理发送CONNACK消息以响应连接时，将调用此方法
    // obj：mosquitto_new中提供的用户数据
    // rc 连接响应的返回码，其中有：
    //      0-成功
    //      1-连接被拒绝（协议版本不可接受）
    //      2-连接被拒绝（标识符被拒绝）
    //      3-连接被拒绝（经纪人不可用）
    //      4-255-保留供将来使用
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    // 消息回调函数，收到订阅的消息后调用
    // mosq： 有效的mosquitto实例，客户端。
    // on_message 回调函数，格式如下:void callback（struct mosquitto *mosq，void *obj，const struct mosquitto_message *message）
    //     回调的参数：
    //     ①mosq：进行回调的mosquitto实例
    //     ②obj： mosquitto_new中提供的用户数据
    //     ③message: 消息数据，回调完成后，库将释放此变量和关联的内存，客户应复制其所需要的任何数据。
    //         struct mosquitto_message{
    //         int mid; //消息序号ID
    //         char *topic; //主题
    //         void *payload; //主题内容 ，MQTT 中有效载荷
    //         int payloadlen; //消息的长度，单位是字节
    //         int qos;     //服务质量
    //         bool retain; //是否保留消息
    //         };
    mosquitto_message_callback_set(mosq, my_message_callback);
    // 设置订阅回调。当代理响应订阅请求时，将调用此函数
    // mosq：结构体mosquitto的指针
    // n_subscribe：一个回调函数
    //     void (*on_subscribe)(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
    //         ①mosq：结构体mosquitto的指针
    //         ②obj：创建客户端的回调参数，是mosquitto_new中提供的用户数据
    //         ③mid：订阅消息的消息 id
    //         ④qos_count：授予的订阅数（granted_qos的大小）
    //         ⑤granted_qos：一个整数数组，指示为每个订阅授予的 QoS
    mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);

     //设置密码,  如果不设置密码，将收不到主题数据
    if(mosquitto_username_pw_set(mosq, USERNAME, PASSWORD) != MOSQ_ERR_SUCCESS) {
        printf("client username_pw_set failed..\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    //连接服务器
    if(mosquitto_connect(mosq, HOST, PORT, KEEP_ALIVE)){
        printf("Unable to connect.\n");
        return 1;
    }
    //开启一个线程，在线程里不停的调用 mosquitto_loop() 来处理网络信息
    int loop = mosquitto_loop_start(mosq);
    if(loop != MOSQ_ERR_SUCCESS)
    {
        printf("mosquitto loop error\n");
        return 1;
    }

    while(1)
    {
       sleep(1);
    }

    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();

    return 0;
}

MQTT总结

mqtt环境搭建
发布订阅模式
qos0 1 2时序图
mqtt协议格式
mqtt客户端api
主题设计、通配符问题
mqtt用户安全认证授权问题

2024-11-11 该篇文章被 Cleofwine 归为分类: 服务端