Bối cảnh & Lý do: Vượt xa khỏi HTTP cho IoT
Trong giai đoạn đầu sự nghiệp, tôi đã xây dựng các backend IoT theo cách mà hầu hết các nhà phát triển web vẫn làm: sử dụng các REST API. Nó hoạt động ổn với một vài thiết bị. Tuy nhiên, mọi thứ bắt đầu đổ vỡ khi tôi chạm ngưỡng 100 thiết bị. Các HTTP header rất nặng nề. Khi hàng ngàn cảm biến gửi payload 50-byte sau mỗi vài giây, một header 500-byte sẽ trở thành sự lãng phí băng thông và thời lượng pin khủng khiếp.
MQTT (Message Queuing Telemetry Transport) giải quyết vấn đề này bằng cách duy trì một kết nối nhẹ và liên tục. Có lý do để nó trở thành tiêu chuẩn công nghiệp. Trong khi MQTT thuần có thể nhanh chóng biến thành mớ mã nguồn “spaghetti” khó quản lý, NestJS cung cấp một lớp trừu tượng microservices giúp mọi thứ gọn gàng hơn. Bằng cách sử dụng NestJS, bạn có thể xử lý các thông điệp MQTT như các sự kiện (events) tiêu chuẩn, tách biệt việc tiếp nhận dữ liệu khỏi logic nghiệp vụ cốt lõi.
Việc chuyển sang kiến trúc này trong một dự án gần đây đã giúp giảm 40% mức sử dụng CPU của máy chủ. Hệ thống vẫn phản hồi tốt ngay cả khi lượng thiết bị kết nối tăng vọt trong quá trình triển khai firmware. Hướng dẫn này sẽ đi sâu vào thiết lập chính xác mà tôi sử dụng để xử lý dữ liệu cảm biến tần suất cao mà không gặp chút khó khăn nào.
Cài đặt: Thiết lập môi trường
Bạn cần một message broker trước khi có thể xử lý bất kỳ dữ liệu nào. Mặc dù EMQX rất tuyệt vời cho quy mô lớn, nhưng Eclipse Mosquitto là lựa chọn ưu tiên của tôi cho việc phát triển và các khối lượng công việc vừa phải. Nó cực kỳ nhẹ, thường sử dụng ít hơn 10MB RAM khi tải nhẹ.
1. Chạy Mosquitto với Docker
Docker là cách nhanh nhất để khởi chạy một broker mà không làm xáo trộn máy cục bộ của bạn. Tạo một file docker-compose.yml để khởi chạy dịch vụ:
version: '3.8'
services:
mosquitto:
image: eclipse-mosquitto
container_name: mosquitto_broker
ports:
- "1883:1883" # Cổng MQTT tiêu chuẩn
- "9001:9001" # Cổng WebSockets
volumes:
- ./mosquitto.conf:/mosquitto/config/mosquitto.conf
Để giữ mọi thứ đơn giản cho việc thử nghiệm, hãy sử dụng một file mosquitto.conf cơ bản cho phép các kết nối ẩn danh:
persistence true
allow_anonymous true
listener 1883 0.0.0.0
2. Khởi tạo dự án NestJS
Bắt đầu bằng cách cài đặt NestJS CLI. Tôi khuyên bạn nên chia hệ thống của mình thành một Gateway và một Telemetry Service, nhưng trong bài hướng dẫn này, chúng ta sẽ tập trung vào thiết lập microservice cốt lõi.
# Cài đặt NestJS CLI
npm install -g @nestjs/cli
# Tạo dự án
nest new iot-backend
cd iot-backend
# Cài đặt các gói microservices và MQTT transport
npm install @nestjs/microservices mqtt
Cấu hình: Triển khai Logic Microservice
NestJS coi MQTT như một lớp vận chuyển (transport layer). Bạn có thể xử lý dữ liệu đến bằng các decorator như @MessagePattern hoặc @EventPattern. Đối với telemetry, @EventPattern là lựa chọn tốt hơn vì nó hoạt động theo cơ chế “fire-and-forget” (gửi và quên), hoàn hảo cho các cập nhật cảm biến tốc độ cao.
1. Cấu hình điểm đầu vào Microservice
Sửa đổi main.ts để yêu cầu NestJS lắng nghe các thông điệp MQTT thay vì các yêu cầu HTTP thông thường. Điều này biến ứng dụng của bạn thành một bộ xử lý thông điệp chuyên dụng.
import { NestFactory } from '@nestjs/core';
import { Transport, MicroserviceOptions } from '@nestjs/microservices';
import { AppModule } from './app.module';
async function bootstrap() {
const app = await NestFactory.createMicroservice<MicroserviceOptions>(
AppModule,
{
transport: Transport.MQTT,
options: {
url: 'mqtt://localhost:1883',
// Đối với môi trường production, hãy sử dụng một clientId duy nhất để theo dõi trạng thái phiên (session)
},
},
);
await app.listen();
console.log('Telemetry Microservice đang hoạt động');
}
bootstrap();
2. Xử lý dữ liệu cảm biến gửi đến
Tiếp theo, tạo một controller để lắng nghe một topic cụ thể. Sử dụng ký tự đại diện (wildcard) + để khớp với bất kỳ ID cảm biến nào. Ví dụ: sensors/sensor_01/data và sensors/sensor_02/data đều sẽ kích hoạt cùng một trình xử lý.
import { Controller } from '@nestjs/common';
import { EventPattern, Payload, Ctx, MqttContext } from '@nestjs/microservices';
@Controller()
export class TelemetryController {
@EventPattern('sensors/+/data')
handleTelemetry(@Payload() data: any, @Ctx() context: MqttContext) {
const topic = context.getTopic();
const sensorId = topic.split('/')[1];
// Xử lý tác vụ mất khoảng 10-20ms ở đây, ví dụ như chèn vào cơ sở dữ liệu
console.log(`Đang xử lý Cảm biến [${sensorId}]:`, data);
this.processData(sensorId, data);
}
private processData(id: string, payload: any) {
// Logic nghiệp vụ nằm ở đây
}
}
3. Gửi lệnh đến thiết bị
Giao tiếp trong IoT là con đường hai chiều. Bạn thường cần đẩy các lệnh ngược lại phần cứng, chẳng hạn như cập nhật màn hình hoặc bật/tắt rơ-le. ClientProxy giúp việc này trở nên đơn giản.
import { Injectable, Inject } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
@Injectable()
export class CommandService {
constructor(
@Inject('MQTT_SERVICE') private client: ClientProxy,
) {}
sendCommand(deviceId: string, command: string) {
const pattern = `devices/${deviceId}/commands`;
const payload = { action: command, timestamp: new Date().toISOString() };
return this.client.emit(pattern, payload);
}
}
Kiểm chứng & Mở rộng: Các lưu ý khi triển khai thực tế
Viết code mới chỉ là một nửa chặng đường. Bạn cần đảm bảo hệ thống sống sót qua lưu lượng truy cập thực tế và các sự cố mất kết nối.
1. Quan sát với MQTT Explorer
Đừng đoán mò xem thông điệp của bạn đã được gửi đi hay chưa. Hãy tải MQTT Explorer và kết nối nó với localhost:1883. Nó trực quan hóa cây topic của bạn theo thời gian thực. Nếu dịch vụ NestJS của bạn không phản hồi, hãy kiểm tra công cụ này trước để xem broker có thực sự nhận được các gói tin hay không.
2. Mở rộng quy mô theo chiều ngang và Shared Subscriptions
Một rào cản lớn với MQTT là nhiều instance của một dịch vụ sẽ nhận được một bản sao của cùng một thông điệp. Điều này dẫn đến việc trùng lặp dữ liệu trong database. Để khắc phục, hãy sử dụng Shared Subscriptions. Bằng cách thêm tiền tố $share/group_name/ vào topic của bạn, broker sẽ cân bằng tải (load-balance) các thông điệp giữa tất cả các instance dịch vụ đang hoạt động. Điều này cho phép bạn mở rộng quy mô lên hàng trăm nghìn thông điệp mỗi giây.
3. Chọn mức QoS phù hợp
MQTT cung cấp ba mức Chất lượng Dịch vụ (QoS). Đối với hầu hết dữ liệu telemetry, QoS 1 (At least once) là lựa chọn tối ưu. Nó đảm bảo thông điệp đến được backend nhưng tốn rất ít tài nguyên. NestJS tự động gửi xác nhận (PUBACK) sau khi hàm controller của bạn kết thúc. Nếu mã của bạn bị crash giữa chừng, broker sẽ thử gửi lại, đảm bảo không có dữ liệu quan trọng nào bị mất khi khởi động lại.
Kiến trúc này cung cấp sự phân tách rõ ràng giữa các thành phần. Nó cho phép đội ngũ của bạn tập trung vào xử lý dữ liệu thay vì quản lý heartbeat của socket. Đối với bất kỳ dự án IoT dài hạn nào, sự kết hợp giữa NestJS và MQTT là một nền tảng vững chắc, có thể mở rộng khi số lượng thiết bị của bạn tăng lên.
