EMQX: Enterprise MQTT Broker untuk IoT

1. Apa Itu EMQX?

EMQX adalah MQTT broker open-source yang dirancang untuk skala enterprise. Dikembangkan oleh EMQ Technologies, EMQX mampu menangani 100+ juta koneksi MQTT simultan pada satu cluster, menjadikannya salah satu MQTT broker paling scalable di dunia.

EMQX mendukung MQTT 3.1, 3.1.1, dan 5.0 (termasuk fitur seperti shared subscriptions, message expiry, dan user properties). EMQX juga dilengkapi dengan Rule Engine untuk memproses dan merutekan data, Bridges untuk menghubungkan beberapa broker, dan berbagai opsi autentikasi.

EMQX vs MQTT Broker Lainnya

Fitur Utama

2. Instalasi & Konfigurasi

Docker Installation

# docker-compose.yml
version: "3.8"
services:
  emqx:
    image: emqx/emqx:5.6
    ports:
      - "1883:1883"      # MQTT
      - "8083:8083"      # MQTT over WebSocket
      - "8084:8084"      # MQTT over WSS
      - "8883:8883"      # MQTT over TLS
      - "18083:18083"    # Dashboard & REST API
    volumes:
      - emqx-data:/opt/emqx/data
      - emqx-log:/opt/emqx/log
      - ./emqx.conf:/opt/emqx/etc/emqx.conf
    environment:
      - EMQX_NAME=emqx-node1
      - EMQX_HOST=127.0.0.1
      - EMQX_LOADED_PLUGINS="emqx_dashboard"
      - EMQX_LISTENER__TCP__EXTERNAL__ACCEPTORS=64
      - EMQX_LISTENER__TCP__EXTERNAL__MAX_CONNECTIONS=1000000
    restart: unless-stopped

volumes:
  emqx-data:
  emqx-log:

Konfigurasi Dasar (emqx.conf)

%% emqx.conf

%% Node name
node {
  name = "emqx@127.0.0.1"
  cookie = "emqx-secret-cookie"
  data_dir = "/opt/emqx/data"
}

%% MQTT Listener
listeners.tcp.default {
  bind = "0.0.0.0:1883"
  max_connections = 1000000
  acceptors = 64
}

%% WebSocket Listener
listeners.ws.default {
  bind = "0.0.0.0:8083"
  max_connections = 100000
  path = "/mqtt"
}

%% TLS Listener
listeners.ssl.default {
  bind = "0.0.0.0:8883"
  max_connections = 500000
  ssl_options {
    certfile = "/opt/emqx/etc/certs/server.crt"
    keyfile = "/opt/emqx/etc/certs/server.key"
    cacertfile = "/opt/emqx/etc/certs/ca.crt"
  }
}

%% MQTT Settings
mqtt {
  max_packet_size = 1MB
  max_clientid_len = 256
  max_topic_levels = 10
  max_qos_allowed = 2
  max_topic_alias = 65535
  retain_available = true
  wildcard_subscription = true
  shared_subscription = true
  max_mqueue_len = 1000
  mqueue_store_qos0 = true
}

%% Dashboard
dashboard {
  listeners.http {
    bind = "0.0.0.0:18083"
  }
  default_username = "admin"
  default_password = "public"
}

Verifikasi

# Cek EMQX status
curl -s http://localhost:18083/api/v5/status

# Dapatkan cluster info
curl -s -u admin:public http://localhost:18083/api/v5/stats

# Test MQTT publish
mosquitto_pub -h localhost -p 1883 -t "test/hello" -m "Hello EMQX!"

# Test MQTT subscribe
mosquitto_sub -h localhost -p 1883 -t "test/#"

# Cek connected clients
curl -s -u admin:public http://localhost:18083/api/v5/clients

3. Clustering & High Availability

EMQX mendukung clustering dengan beberapa strategi discovery: manual, DNS-based, dan etcd/K8s-based. Cluster EMQX secara otomatis mendistribusikan koneksi client ke semua node.

Cluster Setup dengan Docker Compose

# docker-compose-cluster.yml
version: "3.8"
services:
  emqx1:
    image: emqx/emqx:5.6
    environment:
      - EMQX_NAME=emqx1
      - EMQX_HOST=node1.emqx.io
      - EMQX_CLUSTER__STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS="emqx1@node1.emqx.io,emqx2@node2.emqx.io,emqx3@node3.emqx.io"
    ports:
      - "1883:1883"
      - "18083:18083"
    networks:
      emqx-net:
        aliases: [node1.emqx.io]

  emqx2:
    image: emqx/emqx:5.6
    environment:
      - EMQX_NAME=emqx2
      - EMQX_HOST=node2.emqx.io
      - EMQX_CLUSTER__STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS="emqx1@node1.emqx.io,emqx2@node2.emqx.io,emqx3@node3.emqx.io"
    ports:
      - "1884:1883"
      - "18084:18083"
    networks:
      emqx-net:
        aliases: [node2.emqx.io]

  emqx3:
    image: emqx/emqx:5.6
    environment:
      - EMQX_NAME=emqx3
      - EMQX_HOST=node3.emqx.io
      - EMQX_CLUSTER__STRATEGY=static
      - EMQX_CLUSTER__STATIC__SEEDS="emqx1@node1.emqx.io,emqx2@node2.emqx.io,emqx3@node3.emqx.io"
    ports:
      - "1885:1883"
      - "18085:18083"
    networks:
      emqx-net:
        aliases: [node3.emqx.io]

  haproxy:
    image: haproxy:2.9
    ports:
      - "1880:1883"
    volumes:
      - ./haproxy.cfg:/usr/local/etc/haproxy/haproxy.cfg:ro
    depends_on: [emqx1, emqx2, emqx3]
    networks: [emqx-net]

networks:
  emqx-net:

Load Balancing dengan HAProxy

# haproxy.cfg
frontend mqtt_frontend
    bind *:1883
    mode tcp
    option tcplog
    default_backend mqtt_backend

backend mqtt_backend
    mode tcp
    balance roundrobin
    server emqx1 emqx1:1883 check send-proxy-v2
    server emqx2 emqx2:1883 check send-proxy-v2
    server emqx3 emqx3:1883 check send-proxy-v2

# HAProxy menggunakan "sticky" berdasarkan client IP atau client ID
# untuk memastikan reconnect ke node yang sama (session persistence)

4. Rule Engine & Data Integration

Rule Engine EMQX memproses MQTT message berdasarkan SQL-like query dan mengirim hasilnya ke berbagai sink (database, HTTP, Kafka, dll). Ini sangat powerful untuk membangun data pipeline IoT.

Membuat Rule via Dashboard

# Rule SQL syntax:
SELECT
  clientid,
  topic,
  payload.temperature AS temp,
  payload.humidity AS humidity,
  timestamp
FROM
  "sensors/+/data"
WHERE
  payload.temperature > 35

# Actions (sink):
# 1. Republish — publish ke topic lain
# 2. Console — log ke console
# 3. HTTP — kirim ke REST API
# 4. Database — simpan ke MySQL/PostgreSQL/MongoDB/Redis
# 5. Kafka — publish ke Kafka topic
# 6. RabbitMQ — publish ke RabbitMQ exchange

Contoh Rule: Simpan ke Database

-- Rule: Simpan semua sensor data ke PostgreSQL
SELECT
  clientid AS device_id,
  payload.temperature AS temperature,
  payload.humidity AS humidity,
  payload.battery AS battery,
  payload.timestamp AS device_timestamp,
  timestamp AS server_timestamp
FROM
  "devices/+/telemetry"

-- Sink: PostgreSQL
-- INSERT INTO sensor_data (device_id, temperature, humidity, battery, device_ts, server_ts)
-- VALUES (${device_id}, ${temperature}, ${humidity}, ${battery}, ${device_timestamp}, ${server_timestamp})

Contoh Rule: Alert ke HTTP Endpoint

-- Rule: Kirim alert ke webhook saat suhu tinggi
SELECT
  clientid AS device_id,
  payload.temperature AS temperature,
  payload.location AS location,
  timestamp
FROM
  "sensors/#"
WHERE
  payload.temperature > 40

-- Sink: HTTP Action
-- URL: https://api.example.com/alerts
-- Method: POST
-- Body: {"device":"${device_id}","temp":${temperature},"location":"${location}","time":${timestamp}}
-- Headers: Content-Type: application/json, Authorization: Bearer xxx

Contoh Rule: Republish ke Topic Lain

-- Rule: Forward processed data ke topic yang berbeda
SELECT
  payload.temperature AS temp,
  payload.humidity AS humidity,
  CASE
    WHEN payload.temperature > 40 THEN 'critical'
    WHEN payload.temperature > 35 THEN 'warning'
    ELSE 'normal'
  END AS severity
FROM
  "raw/sensors/+/data"

-- Sink: Republish
-- Topic: processed/sensors/${clientid}/status
-- QoS: 1
-- Retain: true

5. MQTT Bridges & Federation

EMQX mendukung bridging ke MQTT broker lain, Kafka, RabbitMQ, dan berbagai message queue. Bridge memungkinkan kamu membangun arsitektur multi-broker atau hybrid cloud.

# MQTT Bridge configuration (via dashboard atau config file)
bridges {
  mqtt {
    my_bridge {
      enable = true
      connector {
        server = "remote-emqx:1883"
        clientid = "bridge-node1"
        username = "bridge-user"
        password = "bridge-pass"
      }
      direction = egress  # egress = kirim, ingress = terima, bidirectional = dua arah
      remote_topic = "cloud/sensors/#"
      local_topic = "local/sensors/#"
    }
  }
}

# Kafka Bridge
bridges {
  kafka {
    my_kafka {
      enable = true
      bootstrap_hosts = "kafka1:9092,kafka2:9092"
      topic = "iot-sensor-data"
      connector {
        kafka_headers = "content-type: application/json"
      }
    }
  }
}

6. Authentication & Authorization

EMQX mendukung berbagai metode autentikasi yang bisa dikombinasikan (chain). Setiap koneksi client akan di-autentikasi secara berurutan sampai berhasil atau ditolak.

Authentication Methods

Konfigurasi Built-in Database Auth

# Tambah user via REST API
curl -X POST "http://localhost:18083/api/v5/authentication/password_based:built_in_database/users" \
  -u "admin:public" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "sensor-01",
    "password": "sensor-01-secret"
  }'

# Bulk create users
curl -X POST "http://localhost:18083/api/v5/authentication/password_based:built_in_database/users" \
  -u "admin:public" \
  -H "Content-Type: application/json" \
  -d '[
    {"user_id": "sensor-01", "password": "pass-01"},
    {"user_id": "sensor-02", "password": "pass-02"},
    {"user_id": "sensor-03", "password": "pass-03"}
  ]'

# Authorization (ACL) — batasi topic per user
curl -X POST "http://localhost:18083/api/v5/authorization/sources:built_in_database/rules" \
  -u "admin:public" \
  -H "Content-Type: application/json" \
  -d '{
    "rules": [
      {"user_id": "sensor-01", "permission": "allow", "action": "publish", "topic": "devices/sensor-01/#"},
      {"user_id": "sensor-01", "permission": "allow", "action": "subscribe", "topic": "commands/sensor-01"},
      {"user_id": "sensor-01", "permission": "deny", "action": "publish", "topic": "#"}
    ]
  }'

7. Topic Rewrite & Retained Messages

Topic Rewrite

%% Topic rewrite rules — transform topic sebelum routing
%% Berguna untuk: normalisasi topic, add tenant prefix, version mapping

rewrite {
  rules = [
    {
      action = "publish"
      source_topic = "sensors/+/data"
      dest_topic = "v2/sensors/${1}/telemetry"
      regex = "sensors/(.+)/data"
    }
    {
      action = "subscribe"
      source_topic = "device/#"
      dest_topic = "tenant-a/device/${1}"
      regex = "device/(.+)"
    }
  ]
}

Retained Messages

# Publish retained message (disimpan dan dikirim ke subscriber baru)
mosquitto_pub -h localhost -p 1883 \
  -t "devices/sensor-01/status" \
  -m '{"online": true, "last_seen": 1719600000}' \
  -r  # retain flag

# Subscriber baru yang subscribe ke topic ini akan langsung
# menerima retained message tanpa menunggu publish baru

# Cek retained messages via API
curl -s -u admin:public "http://localhost:18083/api/v5/retained/messages"

# Hapus retained message
mosquitto_pub -h localhost -p 1883 \
  -t "devices/sensor-01/status" \
  -m "" \
  -r

8. Monitoring & Management

# EMQX Dashboard: http://localhost:18083
# Default: admin / public

# REST API — Monitoring endpoints
# Stats
curl -s -u admin:public http://localhost:18083/api/v5/stats
# {"connections.count": 1523, "topics.count": 456, ...}

# Active clients
curl -s -u admin:public "http://localhost:18083/api/v5/clients?limit=100"

# Subscriptions
curl -s -u admin:public "http://localhost:18083/api/v5/subscriptions"

# Topics
curl -s -u admin:public "http://localhost:18083/api/v5/topics"

# Metrics
curl -s -u admin:public http://localhost:18083/api/v5/metrics
# {"messages.received": 1234567, "messages.delivered": 1234500, ...}

# Prometheus metrics (built-in)
curl http://localhost:18083/api/v5/prometheus/stats

9. Quiz: Uji Pemahamanmu!

Setelah membaca tutorial di atas, jawablah 5 pertanyaan berikut untuk menguji pemahamanmu tentang EMQX: