paho.mqtt.embedded-c-master\MQTTPacket\samples，这个目录里面封装了发布消息、订阅消息的示例。运行pub0sub1，这个示例里面会去订阅主题消息、发布主题消息。并且订阅和发布的消息是同一个主题，所以在运行过程中会看到循环打印同一份消息。代码如下：
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include "MQTTPacket.h"
#include "transport.h"
#include <signal.h>
int toStop = 0;
void cfinish(int sig)
{
signal(SIGINT, NULL);
toStop = 1;
}
void stop_init(void)
{
signal(SIGINT, cfinish);
signal(SIGTERM, cfinish);
}
int main(int argc, char *argv[])
{
MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
int rc = 0;
int mysock = 0;
unsigned char buf[200];
int buflen = sizeof(buf);
int msgid = 1;
MQTTString topicString = MQTTString_initializer;
int req_qos = 0;
char* payload = "mypayload";
int payloadlen = strlen(payload);
int len = 0;
char *host = "m2m.eclipse.org";
int port = 1883;
stop_init();
if (argc > 1)
host = argv[1];
if (argc > 2)
port = atoi(argv[2]);
mysock = transport_open(host, port);
if(mysock < 0)
return mysock;
printf("Sending to hostname %s port %d\n", host, port);
data.clientID.cstring = "me";
data.keepAliveInterval = 20;
data.cleansession = 1;
data.username.cstring = "testuser";
data.password.cstring = "testpassword";
len = MQTTSerialize_connect(buf, buflen, &data);
rc = transport_sendPacketBuffer(mysock, buf, len);
if (MQTTPacket_read(buf, buflen, transport_getdata) == CONNACK)
{
unsigned char sessionPresent, connack_rc;
if (MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, buflen) != 1 || connack_rc != 0)
{
printf("Unable to connect, return code %d\n", connack_rc);
goto exit;
}
}
else
goto exit;
topicString.cstring = "substopic";
len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
rc = transport_sendPacketBuffer(mysock, buf, len);
if (MQTTPacket_read(buf, buflen, transport_getdata) == SUBACK)
{
unsigned short submsgid;
int subcount;
int granted_qos;
rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);
if (granted_qos != 0)
{
printf("granted qos != 0, %d\n", granted_qos);
goto exit;
}
}
else
goto exit;
topicString.cstring = "pubtopic";
while (!toStop)
{
if (MQTTPacket_read(buf, buflen, transport_getdata) == PUBLISH)
{
unsigned char dup;
int qos;
unsigned char retained;
unsigned short msgid;
int payloadlen_in;
unsigned char* payload_in;
int rc;
MQTTString receivedTopic;
rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,
&payload_in, &payloadlen_in, buf, buflen);
printf("message arrived %.*s\n", payloadlen_in, payload_in);
}
printf("publishing reading\n");
len = MQTTSerialize_publish(buf, buflen, 0, 0, 0, 0, topicString, (unsigned char*)payload, payloadlen);
rc = transport_sendPacketBuffer(mysock, buf, len);
}
printf("disconnecting\n");
len = MQTTSerialize_disconnect(buf, buflen);
rc = transport_sendPacketBuffer(mysock, buf, len);
exit:
transport_close(mysock);
return 0;
}
在这个示例中，改造一下，把发布消息的代码给注释掉。我们发现，过了一会，循环读消息的地方就再也接收不到服务器发送过来的消息了。推测是此时客户端与服务器之间的长连接已经断了。回到代码，
//设置心跳包间隔时间
data.keepAliveInterval = 20;
Keep Alive timer
The Keep Alive timer is present in the variable header of a MQTT CONNECT message.
The Keep Alive timer, measured in seconds, defines the maximum time interval between >messages received from a client. It enables the server to detect that the network connection to a client has dropped, without having to wait for the long TCP/IP timeout. The client has a responsibility to send a message within each Keep Alive time period. In the absence of a data-related message during the time period, the client sends a PINGREQ message, which the server acknowledges with a PINGRESP message.
If the server does not receive a message from the client within one and a half times the Keep Alive time period (the client is allowed “grace” of half a time period), it disconnects the client as if the client had sent a DISCONNECT message. This action does not impact any of the client’s subscriptions. See DISCONNECT for more details.
If a client does not receive a PINGRESP message within a Keep Alive time period after sending a PINGREQ, it should close the TCP/IP socket connection.
The Keep Alive timer is a 16-bit value that represents the number of seconds for the time period. The actual value is application-specific, but a typical value is a few minutes. The maximum value is approximately 18 hours. A value of zero (0) means the client is not disconnected.
The format of the Keep Alive timer is shown in the table below. The ordering of the 2 bytes of the Keep Alive Timer is MSB, then LSB (big-endian).
pub0sub1示例代码中，删除发布消息代码块只保留订阅消息功能以后，客户端在20秒之内都可以接收到服务器推送过来的消息。如果20秒内客户端没有向服务器发送PINGREQ消息，那么服务器会关闭掉TCP/IP端口连接。
因此，如果希望开启一个可以永远保持订阅消息的客户端，需要在设置的心跳间隔时间内向服务器发送PINGREQ消息。具体做法可以这样，代码如下：
while (!toStop)
{
/****循环订阅消息*************/
topicString.cstring = "pubtopic";
len = MQTTSerialize_subscribe(buf, buflen, 0, msgid, 1, &topicString, &req_qos);
rc = transport_sendPacketBuffer(mysock, buf, len);
printf("publishing reading000\n");
if (MQTTPacket_read(buf, buflen, transport_getdata) == SUBACK)
{
unsigned short submsgid;
int subcount;
int granted_qos;
rc = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, buflen);
if (granted_qos != 0)
{
printf("granted qos != 0, %d\n", granted_qos);
goto exit;
}
}
else
goto exit;
/*****************/
int state = MQTTPacket_read(buf, buflen, transport_getdata);
if (state != -1){
printf("state2 is = %d\n", state);
}
if (state == PUBLISH)
{
unsigned char dup;
int qos;
unsigned char retained;
unsigned short msgid;
int payloadlen_in;
unsigned char* payload_in;
int rc;
MQTTString receivedTopic;
rc = MQTTDeserialize_publish(&dup, &qos, &retained, &msgid, &receivedTopic,
&payload_in, &payloadlen_in, buf, buflen);
printf("message arrived %.*s\n", payloadlen_in, payload_in);
}
}
我们在循环读取消息的代码块中加入了订阅消息，这样可以保持客户端与服务器之间长连接不会断开。当然，这么做效果并不好，因为循环发送订阅消息会对服务器产生比较多的负载。可以对代码做个优化，比如每间隔10秒钟发送一次订阅消息的请求。百度云官方提供的客户端代码就是这么实现的，下一次我们拿出来做对比。文档代码点这里
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
