Emqx: пропустить много сообщений, выдвигаемых протоколом mqtt - PullRequest
Новая
       98

Emqx: пропустить много сообщений, выдвигаемых протоколом mqtt

0 голосов
/ 25 сентября 2019

Я использую протокол mqtt для отправки некоторых сообщений, и мой брокер - emqx. Я пишу этот скрипт с помощью nodejs для отправки 200 000 сообщений по определенной теме: 

const mqtt = require('mqtt');

const options = {
    clientId: "tazikpush",
    clean: true
}
const client = mqtt.connect("mqtt://xxxxxxxxxx", options);
var topic = "/ApplyData/";
var pushOptions = {
    retain: false,
    qos: 2
}; 
const snooze = ms => new Promise(resolve => setTimeout(resolve, ms));

const example = async () => {
    console.log('Waiting 5 sec and then start');
    await snooze(5000);
    for (var i = 0; i < 200000; i++) {
        // await snooze(250);
        client.publish(topic, message, pushOptions);
        console.log(`done! ${i}`);
    }
};
example();

, а с помощью nodejs я пишу подписчика для прослушивания.по этой теме, а затем сохранить данные в базе данных Redis.Но у меня есть проблема о:

Почему слушатель должен остановиться, пока push-клиент не отправит все 200000 сообщений?

Почему подписчик просто получает 100 сообщений?и другое сообщение сброшено.

На моем подписчике я создаю файл js. В этом файле я создаю клиента и подписываюсь на мою тему с помощью qos 2 mqttClient.js 

const mqtt = require('mqtt');
const log4js = require('log4js');
const config = require('config');

const topic_sub = "/ApplyData/";
log4js.configure(JSON.parse(JSON.stringify(config.get('Logger'))));
var logger = log4js.getLogger('app');
logger.level = 'debug';


const options = {
    clientId: "mqttjs01",
    clean: true
}

const client = mqtt.connect("mqtt://xxxxxxx", options);

client.on("connect", () => {
    console.log("connected  " + client.connected);
    client.subscribe(topic_sub, { qos: 2 });
});

client.on("error", (error) => {
    console.log("Can't connect" + error);
    logger.debug(`Client Error : `, error);
});


module.exports = client;

И я использую клиентское событие на моем контроллере.мой подписчик на самом деле рабочий, значит, его работа - это просто подписать сообщение и сохранить эти сообщения и сохранить в базе данных.

В App.js мне требуются: 

const client = require('./mqttClient');
const controller = require('./controller/mainController');

и в главном контроллере я подписываю сообщения вызовом client.on: 

client.on('message', async (topic, message, packet) => {
        console.log(topic);
        if (topic === '/ApplyData/') {
            var jobject = JSON.parse(message);
            jobject.nid = uuid()
            try {
                let res = await redis_cache.cache(jobject);
            } catch (err) {
                console.log(err);
            }

Я отлаживаю брокер с запуском изконсоль после нажатия 200 000 сообщений я получил 

2019-09-25 11:41:50.885 [warning] tazikpush [Session] Dropped qos2 packet 36998 for too many awaiting_rel
2019-09-25 11:41:50.900 [warning] tazikpush [Session] Dropped qos2 packet 36999 for too many awaiting_rel
2019-09-25 11:41:50.900 [warning] tazikpush [Session] Dropped qos2 packet 37000 for too many awaiting_rel
2019-09-25 11:41:50.900 [warning] tazikpush [Session] Dropped qos2 packet 37001 for too many awaiting_rel
2019-09-25 11:41:50.900 [warning] tazikpush [Session] Dropped qos2 packet 37002 for too many awaiting_rel.... 
2019-09-25 11:49:57.544 [warning] tazikpush [Session] Dropped qos2 packet 40292 for too many awaiting_rel
2019-09-25 11:49:57.544 [warning] tazikpush [Session] The PUBREL PacketId 36898 is not found
2019-09-25 11:49:57.544 [warning] tazikpush [Session] The PUBREL PacketId 36899 is not found
2019-09-25 11:49:57.544 [warning] tazikpush [Session] The PUBREL PacketId 36900 is not found
2019-09-25 11:49:57.544 [warning] tazikpush [Session] The PUBREL PacketId 36901 is not found ...

log

1 Ответ

2 голосов
/ 27 сентября 2019

Ваша публикация слишком быстрая, а использование одного клиента для обработки потребления слишком медленное.

Это может быть связано с вашей конфигурацией:

etc/emqx.conf


## Maximum QoS2 packets (Client -> Broker) awaiting PUBREL, 0 means no limit.
##
## Value: Number
zone.external.max_awaiting_rel = 100

## The QoS2 messages (Client -> Broker) will be dropped if awaiting PUBREL timeout.
##
## Value: Duration
zone.external.await_rel_timeout = 300s

Лучший способ - использовать EMQ X Enterprise или поделиться подписками, чтобы добавить больше клиентов:

https://docs.emqx.io/tutorial/v3/en/advanced/share_subscribe.html

...