Introducción
Esta documentación proporciona una guía integral para integrar Node.js con la nuestraMQTT broker(CrystalMQ) o cualquier corredor de MQTT de su elección. Cubre tareas esenciales como establecer conexiones, suscribir temas, sin suscribirse, y intercambiando mensajes. Al seguir estos pasos, usted puede implementar eficazmenteMQTTcomunicación dentro Aplicaciones Node.js.
Pre-requisitos
Antes de empezar, asegúrese de que tiene lo siguiente:
- Node JS- Asegúrate de instalar Node JS con la 18 o superior.
- MQTT Biblioteca
npm Instalación mqtt
MQTT Conectar
Esta sección cubre varias maneras de conectarse a un corredor MQTT usando Node JS. Antes de proceder, asegúrese de que su corredor soporta el método de conexión deseado (TCP, TLS, WebSocket, etc.) y tiene los parámetros de conexión correctos:
- Broker Address
- Puerto
- Nombre de usuario/Password(si es necesario)
- Certificado(para conexiones TLS)
Basic TCP Conexión
const mqtt = require('mqtt');
const broker Url = 'mqt://broker-url:1883';
const conn_params = {
cliente Id: "testclient1",
};
const client = mqtt.connect(brokerUrl, conn_params);
cliente.on('connect', () = {
consola.log('Connected to MQTT broker');
});Conectarse con la autenticación
Para conectarse a un MQTT Broker que requiere Authentication, configure Nombre de usuario & Contraseña en conexión parámetros.
const conn_params = {
cliente Id: "testclient1",
nombre de usuario: "nombre de usuario",
contraseña: "password",
};Conexión sobre TLS/SSL
Para conectarse de forma segura utilizando TLS/SSL:
const fs = require('fs');
const conn_params = {
protocolo: 'mqts',
anfitrión: 'localhost',
puerto: 8883,
cliente Id: "testclient1",
ca: [fs.readFileSync('/path/to/root.crt')],
};Conectar a través de WebSocket
Para conectarse a través de WebSocket en lugar de TCP, mantenga la url del corredor de la manera siguiente:
const broker Url = 'ws://broker-url:10443/mqtt';
MQTT Publish
Para enviar datos a un tema:
const publishInterval = 2000;
const topic = 'testtopic';
cliente.on('connect', () = {
consola.log('Connected to MQTT broker');
const message = JSON.stringify({"temperature": 27.6, "humidity": 34})
setInterval((() = título {}
cliente. public(topic, message, (err) = {}
si (err) {
consola.error('Error editorial message:', err);
♫ ... {
consola.log('Message published:', message);
}
});
}, publicarInterval);
});MQTT Suscríbete
Para recibir mensajes, suscribirse a un filtro de tema. El corredor enviará mensajes iguales a su cliente.
cliente.on('connect', () = {
consola.log('Connected to MQTT broker');
cliente.subscribe('some_topic', (err) = {
si (err) {
consola.error('Error de suscripción:', err);
♫ ... {
consola.log(`Suscribido al tema: ${topic}`);
}
});
});
cliente.on('message', (topic, message) = {}
Inténtalo.
const data = JSON.parse(message.toString()); // parse JSON if applicable
consola.log(`Recibido en '${topic}':`, data);
} catch (e) {
consola.log(`Recibido mensaje no-JSON en '${topic}':`, message.toString());
}
});
cliente.unsubscribe('testtopic', (err) = {
si (err) {
consola.error('Unsubscribe error:', err);
♫ ... {
consola.log('Unsubscribed from topic: testtopic');
}
});Desconexión MQTT
Desconectarse siempre limpiamente para evitar las fugas de recursos y garantizar una sesión adecuada manejo en el corredor:
cliente.end() = Conf {}
consola.log('Disconnected from MQTT broker');
});Cliente MQTTv311
const mqtt = require('mqtt');
const fs = require('fs');
// Toggle Authentication (set to "ENABLED" or "DISABLED")
const AUTH = "ENABLED";
// ---------------- Configuración de carpetas MQTT ---------
const broker Url = 'mqt://broker-url:1883'; // Reemplazar con su URL/IP de broker real
// Parámetros de conexión MQTT
const conn_params = {
cliente Id: "testclient1", // Unique client ID
// Uncomment debajo de las líneas si el corredor requiere autenticación
// nombre de usuario: "nombre de usuario",
// contraseña: "password",
// Uncomment below lines for TLS/SSL secured connection
// protocolo: 'mqts',
// host: 'broker-url',
puerto: 8883,
// ca: [fs.readFileSync('/path/to/root.crt')], / certificado CA de raíz
};
dejar cliente;
// ---------------- Iniciación del cliente MQTT -------
si (AUTH === "ENABLED") {}
cliente = mqtt.connect(brokerUrl, conn_params); // Con URL y opciones
♫ ... {
cliente = mqtt.connect(conn_params); // Sin URL de corredor
}
// ---------------- MQTT Tema Publish Interval...--------------
const topic = 'testtopic'; // Tema para publicar y suscribir
const publishInterval = 2000; // Publicar cada 2 segundos
// -------------- En la conexión exitosa -----------
cliente.on('connect', () = {
consola.log('Connected to MQTT broker');
// Suscribirse al tema
cliente.subscribe(topic, (err) = {}
si (err) {
consola.error('Error de suscripción:', err);
♫ ... {
consola.log(`Suscribido al tema: ${topic}`);
}
});
// Muestra JSON payload para publicar
mensaje const = JSON.stringify({ temperatura: 27.6, humedad: 34 });
// Publicar el mensaje periódicamente
setInterval((() = título {}
cliente. public(topic, message, { keep: true }, (err) = {}
si (err) {
consola.error('Error editorial message:', err);
♫ ... {
consola.log('Message published:', message);
}
});
}, publicarInterval);
});
// ---------------- Mensaje recibido -----
cliente.on('message', (topic, message) = {}
Inténtalo.
// Tratar el mensaje como JSON
const data = JSON.parse(message.toString());
consola.log(`Recibido en '${topic}':`, data);
} catch (e) {
// Fallback si el mensaje no es JSON
consola.log(`Recibido mensaje no-JSON en '${topic}':`, message.toString());
}
});
// ---------------- Llamamientos adicionales para eventos...
cliente.on('error', (err) = {}
consuelo.error('Error:', err);
});
cliente.on('close', () = {}
consola.log('Disconnected from MQTT broker');
});
cliente.on('reconnect', () = {}
consola.log('Reconectarse un corredor de MQTT...');
});
cliente.on('offline', () = {
consola.warn('MQTT cliente está fuera de línea);
});
// ------------ Desinscribir " Desconectar ---------
// Describirse del tema
// cliente.unsubscribe('testtopic', (err) = {}
// si (err) {
// consola.error('Unsubscribe error:', err);
- Si no...
// consola.log('Unsubscribed from topic: testtopic');
// }
});
// Desconectar al cliente
// client.end() = {}
// consola.log('Desconectado de MQTT broker');
});Muestra MQTTv5 Cliente
Asegurar una terminación adecuada de la conexión de su cliente con el corredor para evitar problemas y filtraciones de recursos en ambas partes, manteniendo así la estabilidad del sistema.
Utilice el siguiente código para desconectar al cliente del broker:
const mqtt = require('mqtt');
const fs = require('fs');
// Toggle Authentication (set to "ENABLED" or "DISABLED")
const AUTH = "ENABLED";
// ---------------- Configuración de carpetas MQTT ---------
const broker Url = 'mqt://broker-url:1883'; // Reemplazar con su URL o IP de broker real
// MQTT v5 parámetros de conexión
const conn_params = {
cliente Id: "testclient1", // Unique client ID
protocoloVersión: 5, // Use MQTT 5 versión
limpio: verdadero, // Iniciar una sesión limpia
// Uncomment si se necesita autenticación
// nombre de usuario: "nombre de usuario",
// contraseña: "password",
// Uncomment for TLS connection
// protocolo: 'mqts',
// host: 'broker-url',
puerto: 8883,
// ca: [fs.readFileSync('/path/to/root.crt')],
};
dejar cliente;
// ---------------- Iniciación del cliente MQTT -------
si (AUTH === "ENABLED") {}
cliente = mqtt.connect(brokerUrl, conn_params);
♫ ... {
cliente = mqtt.connect(conn_params);
}
// ---------------- MQTT Tema Publish Interval...--------------
const topic = 'testtopic';
const publishInterval = 2000; // Publicar cada 2 segundos
// -------------- En la conexión exitosa -----------
cliente.on('connect', (packet) = {}
consola.log('Connected to MQTT broker using MQTT v5');
// Suscribirse con opciones v5 (opcional)
cliente.subscribe(topic, { qos: 0 }, (err, granted) = {}
si (err) {
consola.error('Error de suscripción:', err);
♫ ... {
consola.log(`Suscribido al tema: ${topic}`, granted);
}
});
// Carga para publicar
mensaje const = JSON.stringify({ temperatura: 27.6, humedad: 34 });
// Publicar periódicamente
setInterval((() = título {}
cliente. public(topic, message, {
conservar: verdadero,
propiedades:
Mensaje Intervalo de gastos: 60, // El mensaje expira después de 60 segundos ( característica MQTT v5)
}
}, (err) = confianza {}
si (err) {
consola.error('Error editorial message:', err);
♫ ... {
consola.log('Message published:', message);
}
});
}, publicarInterval);
});
// ---------------- Mensaje recibido -----
cliente.on('message', (topic, message, packet) = {}
Inténtalo.
const data = JSON.parse(message.toString());
consola.log(`Recibido en '${topic}':`, data);
} catch (e) {
consola.log(`Recibido mensaje no-JSON en '${topic}':`, message.toString());
}
});
// ---------------- Llamamientos adicionales para eventos...
cliente.on('error', (err) = {}
consuelo.error('Error:', err);
});
cliente.on('close', () = {}
consola.log('Disconnected from MQTT broker');
});
cliente.on('reconnect', () = {}
consola.log('Reconectarse un corredor de MQTT...');
});
cliente.on('offline', () = {
consola.warn('MQTT cliente está fuera de línea);
});
// ------------ Describir y desconectar ---------
// cliente.unsubscribe('testtopic', (err) = {}
// si (err) {
// consola.error('Unsubscribe error:', err);
- Si no...
// consola.log('Unsubscribed from topic: testtopic');
// }
});
// client.end() = {}
// consola.log('Desconectado de MQTT broker');
});