5 MQTT 在“ read 0:1, break “后断开重连

发布于 2020-09-02 14:19:03 浏览：2841 订阅该版

MQTT 在 访问时一段时间会出现“ read 0:1, break ”会断开连接一次，重连可正常发送一段
试过onenet的样例，也试过和通讯猫的通讯，都一段时间后会出现这个提示，后断开重连。
```
msh />onenet_upload_cycle
[D/onenet.sample] buffer : {"temperature":8}
msh />[D/onenet.sample] buffer : {"temperature":52}
[D/onenet.sample] buffer : {"temperature":56}
[D/onenet.sample] buffer : {"temperature":56}
[D/onenet.sample] buffer : {"temperature":19}
[D/onenet.sample] buffer : {"temperature":11}
[D/onenet.sample] buffer : {"temperature":51}
**[D/mqtt] net_read 0:1, break!**
[D/onenet.mqtt] Enter mqtt_offline_callback!
[E/onenet.mqtt] onenet publish failed (-1)!
[E/onenet.sample] upload has an error, stop uploading
[D/mqtt] restart!
[D/onenet.mqtt] Enter mqtt_connect_callback!
[D/mqtt] ipv4 address port: 6002
[D/mqtt] HOST = '183.230.40.39'
[I/mqtt] MQTT server connect success.
[D/onenet.mqtt] Enter mqtt_online_callback!
**[D/mqtt] net_read 0:1, break!**
[D/onenet.mqtt] Enter mqtt_offline_callback!
[D/mqtt] restart!
[D/onenet.mqtt] Enter mqtt_connect_callback!
[D/mqtt] ipv4 address port: 6002
[D/mqtt] HOST = '183.230.40.39'
[I/mqtt] MQTT server connect success.
[D/onenet.mqtt] Enter mqtt_online_callback!
 
```

6 个回答

ashysage 2020-09-02

这家伙很懒，什么也没写！

补充遇到问题的代码段，MQTT的部分代码：
```c
static int net_read(MQTTClient *c, unsigned char *buf,  int len, int timeout)
{
    int bytes = 0;
    int rc;

while (bytes < len)
    {

#ifdef MQTT_USING_TLS
        if (c->tls_session)
        {
            rc = mbedtls_client_read(c->tls_session, &buf[bytes], (size_t)(len - bytes));
            if (rc <= 0)
            {
                bytes = -1;
                break;
            }
            else
            {
                bytes += rc;
            }
            goto _continue;
        }
#endif

rc = recv(c->sock, &buf[bytes], (size_t)(len - bytes), MSG_DONTWAIT);

if (rc == -1)
        {
            if (errno != ENOTCONN && errno != ECONNRESET)
            {
                bytes = -1;
                break;
            }
        }
        else
            bytes += rc;

#ifdef MQTT_USING_TLS
_continue:
#endif
        if (bytes >= len)
        {
            break;
        }

if (timeout > 0)
        {
            fd_set readset;
            struct timeval interval;

LOG_D("net_read %d:%d, timeout:%d", bytes, len, timeout);
            timeout  = 0;

interval.tv_sec = 1;
            interval.tv_usec = 0;

FD_ZERO(&readset);
            FD_SET(c->sock, &readset);

select(c->sock + 1, &readset, RT_NULL, RT_NULL, &interval);
        }
        else
        {
            LOG_D("net_read %d:%d, break!", bytes, len);
            break;
        }
    }

return bytes;
}
```

后面这里的 的timeout=0会不会是问题啊
```c
static int MQTTPacket_readPacket(MQTTClient *c)
{
    int rc = PAHO_FAILURE;
    MQTTHeader header = {0};
    int len = 0;
    int rem_len = 0;

/* 1. read the header byte.  This has the packet type in it */
    if (net_read(c, c->readbuf, 1, 0) != 1)
        goto exit;
```

Polarbear 2020-09-14

这家伙很懒，什么也没写！

ashysage 2020-09-18

这家伙很懒，什么也没写！

Polarbear 2021-06-25

这家伙很懒，什么也没写！

sliver_cyh 2021-07-05

这家伙很懒，什么也没写！

杨鸿兆 2024-10-14

这家伙很懒，什么也没写！

请登录后再发布评论，点击登录

追加回复

smratton 2021-03-13

这家伙很懒，什么也没写！

Game7 2021-12-10

不断学习中

疯狂的小车 2023-03-20

这家伙很懒，什么也没写！

杨鸿兆 2024-10-12

这家伙很懒，什么也没写！

疯狂的小车 2024-11-02

这家伙很懒，什么也没写！

杨鸿兆 2024-11-05

这家伙很懒，什么也没写！

请登录后再发布评论，点击登录

追加回复

lx1024 2024-02-29

这家伙很懒，什么也没写！

![Mqtt Publish Error.jpg](https://oss-club.rt-thread.org/uploads/20240229/36c1a296da88562c1e01f13e0865bd07.jpg.webp)

Mosquitto Broker收到Puslish Message报文后，其Message Identifier为0时，Broker会断开TCP连接。

Mosquitto Broker日志：
1709176157: Sending SUBACK to abcd1234
1709176158: Client abcd1234 disconnected due to protocol error.

```c
typedef struct MQTTMessage
{
    enum QoS qos;
    unsigned char retained;
    unsigned char dup;
    unsigned short id;
    void *payload;
    size_t payloadlen;
} MQTTMessage;
```
没有找到设置MQTTMessage结构体中id的语句，在paho_mqtt_publish函数中添加id自增动语句测试后，能正常发布消息，且不会再断开TCP。

```c
int paho_mqtt_publish(MQTTClient *client, enum QoS qos, const char *topic, const char *msg_str)
{
    static unsigned short id = 0;
    MQTTMessage message = {0};

if (qos != QOS1)
    {
        LOG_E("Not support Qos(%d) config, only support Qos(d).", qos, QOS1);
        return PAHO_FAILURE;
    }

if (++id == 0)
    {
        ++id;
    }

message.qos = qos;
    message.retained = 0;
    message.id = id;
    message.payload = (void *)msg_str;
    message.payloadlen = rt_strlen(message.payload);

if (client->isblocking && client->pub_sem)
    {
        if(rt_sem_take(client->pub_sem, 5 * RT_TICK_PER_SECOND) < 0)
        {
            return PAHO_FAILURE;
        }
    }

return MQTTPublish(client, topic, &message);
}
```
在 https://mqtt.p2hp.com/mqtt311 2.3.1 包唯一标识中找到关闭唯一标识的描述：
SUBSCRIBE，UNSUBSCRIBE，PUBLISH（QoS > 0 的时候）控制包必须包含**非零**的唯一标识[MQTT-2.3.1-1]。每次客户端发送上述控制包的时候，必须分配一个未使用过的唯一标识[MQTT-2.3.1-2]。如果一个客户端重新发送一个特别的控制包，必须使用相同的唯一标识符。唯一标识会在客户端收到相应的确认包之后变为可用。例如PUBLIST在QoS1的时候对应PUBACK；在QoS2时对应PUBCOMP。对于SUBSCRIBE和UNSUBSCRIBE对应SUBACK和UNSUBACK[MQTT-2.3.1-3]。服务端发送QoS>0的PUBLISH时，上述内容同样适用。