最近在使用MQTTClient实现一个类似于消息推送的服务，说实话，真没怎么使用过MQTTClient，也不知道这是个啥？

上网了解了一下，发现MQTT功能挺强(牛)大(逼)，既能做即时通讯，又能搞消息推送等功能。

这里我使用的是消息推送服务，通过和服务器端协商，终于能够与服务器连接，并且能够收发消息了。 所以，简单总结了一下，有了这篇文章。

MQTT介绍

• MQTT
  MQTT基于订阅者模型架构，客户端如果互相通信，必须在同一订阅主题下，即都订阅了同一个topic，客户端之间是没办法直接通讯的。订阅模型显而易见的好处是群发消息的话只需要发布到topic，所有订阅了这个topic的客户端就可以接收到消息了。
  发送消息必须发送到某个topic，重点说明的是不管客户端是否订阅了该topic都可以向topic发送了消息，还有如果客户端订阅了该主题，那么自己发送的消息也会接收到。

• MQTT特点

• 使用发布/订阅消息模式，提供一对多的消息发布，解除应用程序耦合。这一点很类似于XMPP，但是MQTT的信息冗余远小于XMPP
• 对负载内容屏蔽的消息传输
  使用TCP/IP提供网络连接。主流的MQTT是基于TCP连接进行数据推送的，但是同样有基于UDP的版本，叫做MQTT-SN。这两种版本由于基于不同的连接方式，优缺点自然也就各有不同了
• 三种消息传输方式QoS：
  
  • 0代表“至多一次”，消息发布完全依赖底层 TCP/IP 网络。会发生消息丢失或重复。这一级别可用于如下情况，环境传感器数据，丢失一次读记录无所谓，因为不久后还会有第二次发送。
  • 1代表“至少一次”，确保消息到达，但消息重复可能会发生。
  • 2代表“只有一次”，确保消息到达一次。这一级别可用于如下情况，在计费系统中，消息重复或丢失会导致不正确的结果。 （备注：由于服务端采用Mosca实现，Mosca目前只支持到QoS 1）

如果发送的是临时的消息，例如给某topic所有在线的设备发送一条消息，丢失的话也无所谓，0就可以了（客户端登录的时候要指明支持的QoS级别，同时发送消息的时候也要指明这条消息支持的QoS级别），如果需要客户端保证能接收消息，需要指定QoS为1，如果同时需要加入客户端不在线也要能接收到消息，那么客户端登录的时候要指定session的有效性，接收离线消息需要指定服务端要保留客户端的session状态。

具体MQTT的详细介绍可以戳这里 https://baike.baidu.com/item/MQTT/3618851?fr=aladdin

MQTTClient的使用

iOS 环境下开发 MQTT 客户端程序，一般依赖稳定的第三方 FrameWork，由于涉及网络数据传输，建议选择 Object-c 原生的框架，比如 MQTT-Client-Framework。
现在一般常用的有两个MQTT
1) MQTTKit
2) MQTTClient
不过MQTTKit貌似很长时间不维护了， 使用较多的是MQTTClient。

• 集成MQTTClient
  MQTT-Client-Framework
  
  • 用cocopod直接， pod ‘MQTTClient’
  • GitHub下载，把相对应的文件夹拖进工程即可

MQTT-Client-FrameWork 包提供的客户端类有 MQTTSession 和 MQTTSessionManager，建议使用后者维持静态资源，而且已经封装好自动重连等逻辑。初始化时需要传入相关的网络参数。

我使用的是第二种， 引入 #import "MQTTClient.h" #import "MQTTSessionManager.h" 头文件， 遵循 MQTTSessionManagerDelegate协议

使用步骤：
- 建立连接
和服务器端确定好MQTT服务器地址，端口号， 用户名， 密码， 订阅主题topic

/**
      host: 服务器地址
      port: 服务器端口
      tls:  是否使用tls协议，mosca是支持tls的，如果使用了要设置成true
      keepalive: 心跳时间，单位秒，每隔固定时间发送心跳包, 心跳间隔不得大于120s
      clean: session是否清除，这个需要注意，如果是false，代表保持登录，如果客户端离线了再次登录就可以接收到离线消息
      auth: 是否使用登录验证
      user: 用户名
      pass: 密码
      willTopic: 订阅主题
      willMsg: 自定义的离线消息
      willQos: 接收离线消息的级别
      clientId: 客户端id，需要特别指出的是这个id需要全局唯一，因为服务端是根据这个来区分不同的客户端的，默认情况下一个id登录后，假如有另外的连接以这个id登录，上一个连接会被踢下线, 我使用的设备UUID
*/
NSString *clientId = [UIDevice currentDevice].identifierForVendor.UUIDString;
MQTTSessionManager *sessionManager = [[MQTTSessionManager alloc] init];
[sessionManager connectTo:@"121.199.19.126"  
                       port:1883
                        tls:false
                  keepalive:60  //心跳间隔不得大于120s
                      clean:true
                       auth:true
                       user:@"guest"
                       pass:@"guest"
                       will:false
                  willTopic:nil
                    willMsg:nil
                    willQos:0
             willRetainFlag:false
               withClientId:clientId];
    sessionManager.delegate = self;
    self.sessionManager = sessionManager;
 

 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

• 监控连接状态
  连接当前状态，添加对应的回调接口，可以进行相关的业务逻辑处理。

// 添加监听状态观察者
[self.sessionManager addObserver:self
                   forKeyPath:@"state"
                      options:NSKeyValueObservingOptionInitial | NSKeyValueObservingOptionNew
                      context:nil];
 

 
1
2
3
4
5

监听连接状态，进行相应处理。

 // 监听当前连接状态
- (void)observeValueForKeyPath:(NSString *)keyPath ofObject:(id)object change:(NSDictionary *)change context:(void *)context {
    switch (self.sessionManager.state) {
        case MQTTSessionManagerStateClosed:
            NSLog(@"连接已经关闭");
            break;
        case MQTTSessionManagerStateClosing:
            NSLog(@"连接正在关闭");

            break;
        case MQTTSessionManagerStateConnected:
            NSLog(@"已经连接");

            break;
        case MQTTSessionManagerStateConnecting:
            NSLog(@"正在连接中");

            break;
        case MQTTSessionManagerStateError: {
            NSString *errorCode = self.sessionManager.lastErrorCode.localizedDescription;
            NSLog(@"连接异常 ----- %@",errorCode);
        }

            break;
        case MQTTSessionManagerStateStarting:
            NSLog(@"开始连接");
           break;
        default:
            break;
    }
}

 

 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

• 接收消息
  实现MQTTSessionManagerDelegate代理方法，处理数据。

// 获取服务器返回数据
- (void)handleMessage:(NSData *)data onTopic:(NSString *)topic retained:(BOOL)retained {
    NSLog(@"------------->>%@",topic);

    NSString *dataString = [[NSString alloc] initWithData:data encoding:NSUTF8StringEncoding];
    NSLog(@"%@",dataString);

     // 进行消息处理
}

 

 
1
2
3
4
5
6
7
8
9
10

• 订阅和发送消息

// 订阅主题    NSDictionary类型，Object 为 QoS，key 为 Topic
self.sessionManager.subscriptions = [NSDictionary dictionaryWithObject:[NSNumber numberWithInt:MQTTQosLevelExactlyOnce] forKey:@"hello"];

// 发送消息   返回值msgid大于0代表发送成功
NSString *msg = @"hahaha";
UInt16 msgid = [self.sessionManager sendData:[msg dataUsingEncoding:NSUTF8StringEncoding] //要发送的消息体
                                      topic:@"hello" //要往哪个topic发送消息
                                        qos:0 //消息级别
                                     retain:false];

 

 
1
2
3
4
5
6
7
8
9
10

如果是使用阿里云的服务器替代自己的服务器，需要在阿里云控制台申请 Topic，Group ID 等资源。
在建立连接时，传入的参数值也会有所改变， user 和 pass 由于服务端需要对客户端进行鉴权，因此需要传入合法的 user 和 pass。 user 设置为当前用户的 AccessKey，pass 则设置为 MQTT 客户端 GroupID 的签名字符串，签名计算方式是使用 SecretKey 对 GroupID 做 HmacSHA1 散列加密。

self.manager = [[MQTTSessionManager alloc] init];
self.manager.delegate = self;
self.manager.subscriptions = [NSDictionary dictionaryWithObject:[NSNumber numberWithInt:self.qos]
                                                                 forKey:[NSString stringWithFormat:@"%@/#", self.rootTopic]];
//password的计算方式是，使用secretkey对groupId做hmac签名算法，具体实现参考macSignWithText方法
NSString *passWord = [[self class] macSignWithText:self.groupId secretKey:self.secretKey];
[self.manager connectTo:self.mqttSettings[@"host"]
                                  port:[self.mqttSettings[@"port"] intValue]
                                   tls:[self.mqttSettings[@"tls"] boolValue]
                             keepalive:60  //心跳间隔不得大于120s
                                 clean:true
                                  auth:true
                                  user:self.accessKey
                                  pass:passWord
                                  will:false
                             willTopic:nil
                               willMsg:nil
                               willQos:0
                        willRetainFlag:false
                          withClientId:self.clientId];
 

 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

使用 SecretKey 对 GroupID 做 HmacSHA1 散列加密

+ (NSString *)macSignWithText:(NSString *)text secretKey:(NSString *)secretKey {
    NSData *saltData = [secretKey dataUsingEncoding:NSUTF8StringEncoding];
    NSData *paramData = [text dataUsingEncoding:NSUTF8StringEncoding];
    NSMutableData* hash = [NSMutableData dataWithLength:CC_SHA1_DIGEST_LENGTH ];
    CCHmac(kCCHmacAlgSHA1, saltData.bytes, saltData.length, paramData.bytes, paramData.length, hash.mutableBytes);
    NSString *base64Hash = [hash base64EncodedStringWithOptions:0];
    return base64Hash;
}
 

 
1
2
3
4
5
6
7
8

具体介绍请戳下面帮助链接
MQTT接入环境配置
阿里云接入MTQQ示例
申请MQ资源

参考

---

• iOS MQTT—-MQTTClient实战-看这篇的就够了
• iOS MQTT协议笔记之消息推送
• MQTTClient连接问题解决

MQTT

MQTT是一种轻量级物联网消息推送协议，使用MQTT开发物联网设备将会十分简单方便。

关于MQTT服务器的安装可以看我的这篇文章：搭建MQTT服务器

安装paho.mqtt.c库

安装依赖

sudo apt-get install libssl-dev

安装

git clone https://github.com/eclipse/paho.mqtt.c.git

cd paho.mqtt.c
make
sudo make install

编译及运行

将下面的源代码放到test.c文件。

编译

gcc -o test test.c -l paho-mqtt3c

gcc编译引用的API接口库：参考文章

paho-mqtt3a ：异步API接口的库，API函数都是以MQTTAsync开头，API函数定义在MQTTAsync.h。

paho-mqtt3as ：异步API接口的库，API函数都是以MQTTAsync开头，API函数定义在MQTTAsync.h，这个库默认带有TLS加密功能，同时也需要openssl库支持，才能使用这个库。

paho-mqtt3c ：同步API接口的库，API函数都是以MQTTClient开头，API函数定义在MQTTClient.h。

paho-mqtt3cs ：同步API接口的库，API函数都是以MQTTClient开头，API函数定义在MQTTClient.h，这个库默认带有TLS加密功能，同时也需要openssl库支持，才能使用这个库。

运行

./test

源代码

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

#define MQTT_Uri    "localhost"   				// MQTT服务器的地址和端口号
#define ClientId    "ubuntu16"                  // ClientId需要唯一
#define UserName    "pi"                  		// 用户名
#define PassWord    "12345678"                  // 用户名对应的密码

// 失去连接回调函数
void connect_lost(void *context, char *cause)
{
    printf("Connection lost,The reason: %s \n",cause);
}

// 收到主题信息回调函数
int message_arrived(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Receive topic: %s, message data: \n", topicName);
    printf("%.*s\n", message->payloadlen, (char*)message->payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

// 主题发布成功回调函数
void delivery_complete(void *context, MQTTClient_deliveryToken dt)
{
    printf("publish topic success,token = %d \n", dt);
}

int main(int argc, char* argv[])
{
    // 1、定义一个MQTT客户端结构体指针
    MQTTClient client;

    // 2、创建一个MQTT客户端
    int rc;
    if ((rc = MQTTClient_create(&client, MQTT_Uri, ClientId, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to create client, return code %d\n", rc);
        exit(EXIT_FAILURE);
        goto exit;
    }

    // 3、创建一个MQTT连接配置结构体，并配置其参数
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = UserName;          // 用户名 
    conn_opts.password = PassWord;          // 用户名对应的密码
    conn_opts.keepAliveInterval = 60;       // 心跳时间
    conn_opts.cleansession = 1;             // 清除会话
    
    // 4、设置MQTT连接时的回调函数
    if ((rc = MQTTClient_setCallbacks(client, NULL, connect_lost, message_arrived, delivery_complete)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to set callbacks, return code %d\n", rc);
        rc = EXIT_FAILURE;
        goto destroy_exit;
    }

    // 5、开始连接到MQTT服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
        goto destroy_exit;
    }

    // 6、定义一个主题消息存储结构体
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    char mag_data[] = "I am Ubuntu16.";
    pubmsg.payload = mag_data;
    pubmsg.payloadlen = (int)strlen(mag_data);
    pubmsg.qos = 1;                 // qos等级为1 
    pubmsg.retained = 0;            // 服务器不保留消息
    MQTTClient_deliveryToken token; // 标记MQTT消息的值，用来检查消息是否发送成功

    // 7、发布主题信息
    if ((rc = MQTTClient_publishMessage(client, "pubtest", &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to publish message, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // 8、订阅主题
    if ((rc = MQTTClient_subscribe(client, "subtest", 1)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe, return code %d\n", rc);
        rc = EXIT_FAILURE;
    }

    // 9、等待输入'Q'或'q'退出
    printf("Press Q or q + <Enter> to quit\n\n");
    int ch;
    do
    {
            ch = getchar();
    } while (ch!='Q' && ch != 'q');

    if ((rc = MQTTClient_unsubscribe(client, "ubuntu16_subscribe")) != MQTTCLIENT_SUCCESS)
    {
            printf("Failed to unsubscribe, return code %d\n", rc);
            rc = EXIT_FAILURE;
    }

    // 10、断开连接
    if ((rc = MQTTClient_disconnect(client, 10000)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to disconnect, return code %d\n", rc);
        rc = EXIT_FAILURE;
    }

destroy_exit:
    MQTTClient_destroy(&client);
exit:
    return rc;
}

paho实现MQTTClient发布消息

接下来会用paho开源的一个项目，实现mqtt客户端发布消息，此文主要参考MQTT Client library for C，Paho给出的创建一个客户端有如下类似的步骤：

1、安装

//从github上下载项目
git clone https://github.com/eclipse/paho.mqtt.c.git
//进入文件夹
cd paho.mqtt.c
//编译
make
//安装
sudo make install

2、目录介绍

src：源文件、库文件目录
build:编译过后的动态库及执行文件

2.1、src目录

src中有许多源文件，有发布、订阅、同步、异步文件。

2.2、build目录

build文件是编译过后的执行文件，也是同步、异步、发布、订阅消息等功能

3、实现过程

  1.创建一个客户端对象；
  2.设置连接MQTT服务器的选项；
  3.如果多线程（异步模式）操作被使用则设置回调函数（详见 Asynchronous >vs synchronous client applications）；
  4.订阅客户端需要接收的任意话题；
  5.重复以下操作直到结束：
    a.发布客户端需要的任意信息；
    b.处理所有接收到的信息；
  6.断开客户端连接；
  7.释放客户端使用的所有内存。

4、文件操作

我们直接使用paho自带文件，首先修改src中MQTTClient_publish.c源代码。

4.1、修改信息

//mqtt服务器地址
#define ADDRESS “tcp://m2m.eclipse.org:1883”
//客户端号
#define CLIENTID “ExampleClientPub”
//主题
#define TOPIC “MQTT Examples”
//消息
#define PAYLOAD “Hello World!”

4.2、添加信息

需要添加服务器中用户名和密码

 char *username= "test"; //添加的用户名
 char *password = "test"; //添加的密码

将其写入客户端选项

 conn_opts.username = username; //将用户名写入连接选项中
 conn_opts.password = password; //将密码写入连接选项中

5、源码

源码我做了详尽的解释，其中有许多函数需要自己去上面文件自己查看，下面这是同步发布，不是异步，异步的代码也有，大家可以去看上面的网站

/*******************************************************************************
 * Copyright (c) 2012, 2020 IBM Corp.
 *
 * 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
 *******************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"

#define ADDRESS     "59.110.42.24:1883"
#define CLIENTID    "0bd981c5-a055-4196-8b7f-efb9f7a4d6ac"
#define TOPIC       "test"
#define PAYLOAD     "Hello World!"
#define QOS         1
#define TIMEOUT     10000L

int main(int argc, char* argv[])
{
    //声明mqtt客户端
    MQTTClient client;
    //初始化客户端选项 conn_opts
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    //消息初始化 pubmsg
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;
    char *username = "test";
    char *password = "test";
    if ((rc = MQTTClient_create(&client, ADDRESS, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to create client, return code %d\n", rc);
         exit(EXIT_FAILURE);
    }
    //保持心跳20
    conn_opts.keepAliveInterval = 20;
    //清理会话
    conn_opts.cleansession = 1;
    //客户端的用户名和密码
    conn_opts.username = username;
    conn_opts.password = password;
    //创建连接
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    //消息负载（内容）
    pubmsg.payload = PAYLOAD;
    //消息长度
    pubmsg.payloadlen = (int)strlen(PAYLOAD);
    //消息质量分为0：不重要的消息比如温度，可以多次上传丢失一次没事。1：可能会丢失1次 2：永远不会丢失
    pubmsg.qos = QOS;
    //有true和false 判断消息是否保留
    pubmsg.retained = 0;
    //发布消息 token是消息发布后，传递令牌将返回客户端检查令牌是否已成功传递到其目的地
    if ((rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token)) != MQTTCLIENT_SUCCESS)
    {
         printf("Failed to publish message, return code %d\n", rc);
         exit(EXIT_FAILURE);
    }

    printf("Waiting for up to %d seconds for publication of %s\n"
            "on topic %s for client with ClientID: %s\n",
            (int)(TIMEOUT/1000), PAYLOAD, TOPIC, CLIENTID);
    //阻塞函数，等待消息发布
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);

    if ((rc = MQTTClient_disconnect(client, 10000)) != MQTTCLIENT_SUCCESS)
    	printf("Failed to disconnect, return code %d\n", rc);
    //断开连接
    MQTTClient_destroy(&client);
    return rc;
}

6、实现效果

服务器中发布

mqtt测试软件订阅的消息就能接受到