Я пытаюсь публиковать более 100 сообщений в секунду во встроенном концентраторе событий Azure Iot Hub. Я использую библиотеку libmosquitto 1.6.8. Я использую пакет уровня Free для Azure Iot Hub, я знаю, что существует ограничение на 100 сообщений в секунду. Но это не связано с этим вопросом. Мне не удалось опубликовать даже половину сообщений в AZ Iot Hub.
По сути, у меня есть список из нескольких значений в multimap, которые нужно отправить. Список показателей:
std::multimap< const std::string, std::tuple< const std::string, const std::string, float> > calculatedMetricList;
Я буду перебирать multimap и создавать каждое значение в полезную нагрузку объекта и отправлять его. Это означает, что метод mosquitto_publish будет вызываться несколько раз.
Ниже приведен код для публикации сообщений:
void MosquittoClient::sendDataToUpstreamSystem(){
StatisticalMethod statisticalMethod;
int rc;
MosquittoClient pub_mosq(
"<IoT Hub Name>.azure-devices.net",
"<deviceID>",
"<username>",
"<Password>",
"devices/<deviceID>/messages/events/");
printf("Using MQTT to get data payload from host: %s and on port: %d.\r\n", pub_mosq.get_host(), pub_mosq.get_port());
// init the mosquitto lib
mosquitto_lib_init();
// create the mosquito object
struct mosquitto* mosq = mosquitto_new(pub_mosq.get_deviceID(), false, NULL);
// add callback functions
mosquitto_connect_callback_set(mosq, MosquittoClient::connect_callback);
mosquitto_publish_callback_set(mosq, MosquittoClient::publish_callback);
mosquitto_message_callback_set(mosq, MosquittoClient::on_message);
mosquitto_disconnect_callback_set(mosq, MosquittoClient::on_disconnect_callback);
// set mosquitto username, password and options
mosquitto_username_pw_set(mosq, pub_mosq.get_userName(), pub_mosq.get_password());
// specify the certificate to use
std::ifstream infile(pub_mosq.get_certificate());
bool certExists = infile.good();
infile.close();
if (!certExists)
{
printf("Warning: Could not find file '%s'! The mosquitto loop may fail.\r\n", pub_mosq.get_certificate());
}
printf("Using certificate: %s\r\n", pub_mosq.get_certificate());
mosquitto_tls_set(mosq, pub_mosq.get_certificate(), NULL, NULL, NULL, NULL);
// specify the mqtt version to use
int* option = new int(MQTT_PROTOCOL_V311);
rc = mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, option);
if (rc != MOSQ_ERR_SUCCESS)
{
rc = pub_mosq.mosquitto_error(rc, "Error: opts_set protocol version");
}
else
{
printf("Setting up options OK\r\n");
}
// connect
printf("Connecting...\r\n");
rc = mosquitto_connect_async(mosq, pub_mosq.get_host(), pub_mosq.get_port(), 4);
if (rc != MOSQ_ERR_SUCCESS)
{
rc = pub_mosq.mosquitto_error(rc, NULL);
}
else
{
printf("Connect returned OK\r\n");
rc = mosquitto_loop_start(mosq);
if (rc != MOSQ_ERR_SUCCESS)
{
rc = pub_mosq.mosquitto_error(rc, NULL);
}
else
{
do
{
for (auto itr = Metrics::calculatedMetricList.begin(); itr != Metrics::calculatedMetricList.end(); itr++) {
int msgId = rand();
std::string test1= itr->first;
std::string test2 = std::get<0>(itr->second);
std::string test3= std::get<1>(itr->second); // metric type
float value = std::get<2>(itr->second); // value
DataPayload objectPayload(
75162345,
496523,
test3,
value,
"test",
test1,
"test",
"test",
123,
213,
23
);
objectPayload.setPayload();
std::string dataPayload = objectPayload.getPayload();
//DEBUG
std::cout << "dataPayload: " << dataPayload << std::endl;
//DEBUG
std::cout << "dataPayload Size: " << dataPayload.size() << std::endl;
// once connected, we can publish (send) a Telemetry message
printf("Publishing to topic: %s\r\n", pub_mosq.get_topic());
rc = pub_mosq.publishToTopic(mosq, &msgId, dataPayload.size(), (char *)dataPayload.c_str());
if (rc == MOSQ_ERR_SUCCESS)
{
printf("Publish returned OK\r\n");
}
else
{
rc = pub_mosq.mosquitto_error(rc, NULL);
}
}
} while (rc != MOSQ_ERR_SUCCESS);
}
}
mosquitto_loop_stop(mosq, true);
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();}
Метод публикации:
int MosquittoClient::publishToTopic(struct mosquitto *mosq, int *msgId, int sizeOfData, char *data)
{
return mosquitto_publish(mosq, msgId, p_topic, sizeOfData, data, 1, true);
}
При запуске программы все опубликованные сообщения возвращаются в соответствии с консолью. Но только одно или два сообщения появляются на стороне Azure IoT Hub.
На следующем изображении показан мониторинг Центра Интернета вещей, в то время прошло только одно сообщение. Двойной мониторинг обозревателя устройств
Я перепробовал так много разных решений, но программа не смогла опубликовать все сообщения. Похоже, что метод публикации ожидает завершения первого сообщения, но итерация переходит к следующему сообщению, что приводит к его удалению. Если это причина отброшенных сообщений, как лучше всего упорядочить отправку сообщений? В противном случае, что еще могло вызвать отбрасывание сообщений?
Обновлять
Проблема заключалась в том, что программа не дождалась успешной публикации сообщений в брокере (Azure IoT Hub). Вы узнаете, успешно ли опубликовано сообщение для брокера, если будет возвращено publish_callback.
void MosquittoClient::publish_callback(struct mosquitto* mosq, void* userdata, int mid)
{
printf("Publish OK.\r\n");
}
Решением было приостановить поток перед уничтожением, очистить вызовы и запустить цикл Mosquitto до установления соединения.
sleep(30);
mosquitto_loop_stop(mosq, true);
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();
