提高数据集成稳定性：EMQX Platform 端到端规则调试指南

SELECT
  username,
  clientid,
  payload.vehicle_id as vehicle_id,
  payload.speed as speed,
  payload.location.latitude as latitude,
  payload.location.longitude as longitude,
  payload.safety_features.airbag_deployed as airbag_deployed,
  payload.safety_features.abs_status as abs_status,  timestampFROM
  "devices/#"WHERE payload.speed > '120'

const express = require("express");const app = express();
app.use(express.json());
app.post("/speed", (req, res) => {  const { speed, latitude, longitude } = req.body;  console.log(`Received data: Speed is ${speed} km/h at coordinates (${latitude}, ${longitude}).`);
  res.status(200. send("Data received successfully!");
});const PORT = 3000;
app.listen(PORT, () => {  console.log(`Server is running on port ${PORT}`);
});

services:  postgres:    image: postgres    environment:      POSTGRES_DB: emqx      POSTGRES_USER: admin      POSTGRES_PASSWORD: public    ports:
      - "5432:5432"    volumes:
      - pgdata:/var/lib/postgresql/data

CREATE TABLE car_infos (
  id SERIAL PRIMARY KEY,
  username VARCHAR(255),
  clientid VARCHAR(255),  timestamp TIMESTAMP,
  vehicle_id VARCHAR(255),
  speed INT);

INSERT INTO car_infos (
  username,
  clientid,  timestamp,
  vehicle_id,
  speed
) VALUES (
  ${username},
  ${clientid},
  TO_TIMESTAMP(${timestamp}::bigint / 1000),
  ${vehicle_id},
  ${speed}
);