From HTTP Polling to MQTT Push: Evolving High‑Concurrency IoT Device Access Architecture

1. The real problem is not "whether to use MQTT" but whether the system can handle high‑concurrency device access

In a smart‑park scenario with hundreds of thousands of devices (door locks, HVAC, lighting, sensors, edge gateways), three streams of traffic must be processed: telemetry upload, downstream control, and online‑status/alert events. A naïve HTTP‑polling design works for a few thousand devices but collapses when the count reaches tens of thousands because the request‑response model creates massive empty polls, high connection overhead, and no built‑in session handling.

2. Why MQTT fits the device‑centric use case

MQTT provides long‑lived connections, lightweight headers, a publish‑subscribe model, QoS semantics, persistent sessions, and will messages. These features directly address the four structural contradictions of HTTP polling: excessive empty requests, trade‑off between latency and load, per‑request TCP/TLS overhead, and lack of session awareness.

2.1 Protocol comparison (key dimensions)

Communication model : HTTP – request/response; WebSocket/gRPC – bidirectional stream; MQTT – publish/subscribe.

Protocol overhead : HTTP high; WebSocket/gRPC medium; MQTT low.

Device ecosystem : HTTP very broad; WebSocket medium; gRPC server‑centric; MQTT extremely broad.

Weak‑network suitability : HTTP general; WebSocket general; gRPC general; MQTT good.

QoS semantics : none in HTTP/WebSocket/gRPC (application‑level); built‑in in MQTT.

Offline session support : none in HTTP/WebSocket/gRPC (must be built); built‑in in MQTT.

Will message support : none in HTTP/WebSocket/gRPC; built‑in in MQTT.

Thus, when the core object is a device rather than a browser or internal microservice, MQTT is the natural fit.

3. MQTT’s three essential mechanisms

Topic routing : Devices publish/subscribe using hierarchical topics such as v1/{tenant}/{productKey}/{deviceId}/telemetry, .../command, .../command_reply, and sys/{tenant}/{productKey}/{deviceId}/lifecycle. The topic layout determines permission aggregation, consumption sharding, multi‑tenant isolation, and future version upgrades.

QoS semantics :

QoS 0 – at most once, suitable for high‑frequency telemetry where occasional loss is acceptable.

QoS 1 – at least once, used for alerts, state changes, ordinary commands.

QoS 2 – exactly once, reserved for rare strong‑consistency control scenarios.

QoS is not a reliability switch; business‑level idempotency and transaction handling are still required.

Persistent session & offline messages : When a device disconnects, the broker retains subscriptions and pending messages. Upon reconnection, the session is resumed, avoiding the need for the application layer to rebuild message queues.

4. Full production‑grade architecture – four planes

The system must be split into four logical planes to survive high load and long‑term evolution:

┌────────────────────────────┐
                │        Governance plane    │
                │ ACL / Rate‑limit / Quota   │
                │ Observability / Auditing   │
                └─────────────┬──────────────┘
                              │
   ┌──────────────────────────▼───────────────────────────┐
   │               Access plane (MQTT broker cluster)      │
   │ Device SDK / TLS / Load‑balancer / Shared‑subscription│
   └───────────────────────────────┬───────────────────────┘
                                      │
   ┌───────────────────────────────▼───────────────────────┐
   │               Event plane (Kafka / Pulsar)            │
   │ MQTT → Rule Engine → Kafka → Stream Processing        │
   └───────────────────────────────┬───────────────────────┘
                                      │
   ┌───────────────────────────────▼───────────────────────┐
   │               State plane (Device shadow, MySQL, Redis) │
   │ Device shadow / command state / outbox / audit log      │
   └───────────────────────────────────────────────────────┘

Each plane has a clear responsibility and must avoid “overloading” the lower layers with business logic.

4.1 Access plane responsibilities

Accept massive device connections.

Perform authentication and ACL checks.

Route topics and distribute shared subscriptions.

Manage online sessions, offline messages, and will events.

Mixing heavy business logic here leads to scaling, isolation, and upgrade problems.

4.2 Event plane responsibilities

The MQTT broker is not a durable event store. Upstream telemetry is bridged to Kafka (or Pulsar) via a rule engine. Benefits:

Broker and business consumers are decoupled.

Kafka provides back‑pressure handling and replay capability.

Multiple downstream services can independently consume the same event stream.

4.3 State plane responsibilities

Business services need a persistent view of device state. The state plane stores: device_shadow – latest snapshot. device_online_session – heartbeat and connection status. device_command – command master record. device_command_outbox – pending delivery events. device_event_log – audit log.

Without this plane the system becomes a “black box” that only receives messages.

4.4 Governance plane responsibilities

Multi‑tenant quota enforcement.

Per‑device rate limiting.

Topic ACL management.

Canary releases, certificate rotation, alerting, retry & compensation, fault‑injection drills, upgrade windows.

Missing governance turns a high‑throughput system into an uncontrolled avalanche.

5. Evolution stages – from demo to production

5.1 Stage 1: HTTP polling

Simple request/response flow works for 1 000–5 000 devices but suffers from four problems: massive empty polls, high command latency, no session model, and inaccurate online status.

5.2 Stage 2: MQTT without downstream decoupling

Replacing HTTP with MQTT reduces polling overhead but still couples business services directly to the broker, leading to back‑pressure, lack of replay, and tangled state handling.

5.3 Stage 3: Full four‑plane design

Introduce Kafka as an event bus, separate state services, and enforce governance. The broker only handles connection and routing; all business logic lives downstream.

5.4 Stage 4: Multi‑region edge access

When devices span regions, deploy regional broker clusters, keep session state locally, and funnel critical events to a central stream platform. This reduces RTT and avoids cross‑region traffic spikes.

6. Production‑grade device client (Go example)

package device

import (
    "crypto/tls"
    "encoding/json"
    "errors"
    "fmt"
    "log"
    "math/rand"
    "sync"
    "time"

    mqtt "github.com/eclipse/paho.mqtt.golang"
)

type Telemetry struct {
    Tenant     string            `json:"tenant"`
    ProductKey string            `json:"product_key"`
    DeviceID   string            `json:"device_id"`
    Timestamp  int64             `json:"timestamp"`
    Sequence   uint64            `json:"sequence"`
    Metrics    map[string]float64 `json:"metrics"`
}

type ClientConfig struct {
    BrokerURL        string
    ClientID         string
    Username         string
    Password         string
    Tenant           string
    ProductKey       string
    DeviceID         string
    ConnectTimeout   time.Duration
    MaxReconnectWait time.Duration
}

type OfflineStore interface {
    Push([]byte) error
    PopN(int) ([][]byte, error)
}

type DeviceClient struct {
    cfg          ClientConfig
    client       mqtt.Client
    store        OfflineStore
    mu           sync.Mutex
    connected    bool
    commandTopic string
}

func NewDeviceClient(cfg ClientConfig, store OfflineStore) *DeviceClient {
    return &DeviceClient{
        cfg: cfg,
        store: store,
        commandTopic: fmt.Sprintf("v1/%s/%s/%s/command", cfg.Tenant, cfg.ProductKey, cfg.DeviceID),
    }
}

func (c *DeviceClient) Connect() error {
    opts := mqtt.NewClientOptions()
    opts.AddBroker(c.cfg.BrokerURL)
    opts.SetClientID(c.cfg.ClientID)
    opts.SetUsername(c.cfg.Username)
    opts.SetPassword(c.cfg.Password)
    opts.SetConnectTimeout(c.cfg.ConnectTimeout)
    opts.SetAutoReconnect(true)
    opts.SetCleanSession(false)
    opts.SetTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12})

    lifecycleTopic := fmt.Sprintf("sys/%s/%s/%s/lifecycle", c.cfg.Tenant, c.cfg.ProductKey, c.cfg.DeviceID)
    opts.SetWill(lifecycleTopic, mustJSON(map[string]any{"device_id": c.cfg.DeviceID, "status": "offline", "ts": time.Now().UnixMilli()}), 1, true)

    opts.OnConnect = func(mc mqtt.Client) {
        c.mu.Lock(); c.connected = true; c.mu.Unlock()
        if token := mc.Subscribe(c.commandTopic, 1, c.handleCommand); token.Wait() && token.Error() != nil {
            log.Printf("subscribe command topic failed: %v", token.Error())
            return
        }
        if token := mc.Publish(lifecycleTopic, 1, true, mustJSON(map[string]any{"device_id": c.cfg.DeviceID, "status": "online", "ts": time.Now().UnixMilli()})); token.Wait() && token.Error() != nil {
            log.Printf("publish online lifecycle failed: %v", token.Error())
        }
        if err := c.flushOffline(); err != nil {
            log.Printf("flush offline telemetry failed: %v", err)
        }
    }

    opts.SetConnectionLostHandler(func(_ mqtt.Client, err error) {
        c.mu.Lock(); c.connected = false; c.mu.Unlock()
        log.Printf("mqtt connection lost: %v", err)
    })
    opts.SetReconnectingHandler(func(_ mqtt.Client, _ *mqtt.ClientOptions) {
        wait := time.Duration(rand.Int63n(int64(c.cfg.MaxReconnectWait)))
        time.Sleep(wait)
    })

    client := mqtt.NewClient(opts)
    token := client.Connect()
    if !token.WaitTimeout(c.cfg.ConnectTimeout) {
        return errors.New("mqtt connect timeout")
    }
    if token.Error() != nil {
        return token.Error()
    }
    c.client = client
    return nil
}

func (c *DeviceClient) PublishTelemetry(t Telemetry) error {
    payload, err := json.Marshal(t)
    if err != nil { return err }
    topic := fmt.Sprintf("v1/%s/%s/%s/telemetry", c.cfg.Tenant, c.cfg.ProductKey, c.cfg.DeviceID)
    if !c.isConnected() { return c.store.Push(payload) }
    token := c.client.Publish(topic, 1, false, payload)
    if !token.WaitTimeout(3*time.Second) || token.Error() != nil {
        _ = c.store.Push(payload)
        return fmt.Errorf("publish telemetry failed: %w", token.Error())
    }
    return nil
}

func (c *DeviceClient) handleCommand(_ mqtt.Client, msg mqtt.Message) {
    log.Printf("receive command topic=%s payload=%s", msg.Topic(), string(msg.Payload()))
    replyTopic := fmt.Sprintf("v1/%s/%s/%s/command_reply", c.cfg.Tenant, c.cfg.ProductKey, c.cfg.DeviceID)
    reply := map[string]any{"device_id": c.cfg.DeviceID, "result": "SUCCESS", "ts": time.Now().UnixMilli()}
    token := c.client.Publish(replyTopic, 1, false, mustJSON(reply))
    token.WaitTimeout(2 * time.Second)
}

func (c *DeviceClient) flushOffline() error {
    if c.store == nil || c.client == nil { return nil }
    batch, err := c.store.PopN(100)
    if err != nil { return err }
    topic := fmt.Sprintf("v1/%s/%s/%s/telemetry", c.cfg.Tenant, c.cfg.ProductKey, c.cfg.DeviceID)
    for _, payload := range batch {
        token := c.client.Publish(topic, 1, false, payload)
        if !token.WaitTimeout(2*time.Second) || token.Error() != nil {
            return fmt.Errorf("replay offline payload failed: %w", token.Error())
        }
    }
    return nil
}

func (c *DeviceClient) isConnected() bool { c.mu.Lock(); defer c.mu.Unlock(); return c.connected }

func (c *DeviceClient) Close() { if c.client != nil && c.client.IsConnected() { c.client.Disconnect(250) } }

func mustJSON(v any) []byte { data, err := json.Marshal(v); if err != nil { panic(err) }; return data }

The code demonstrates four production concerns: CleanSession=false for session persistence, will messages for graceful offline detection, local offline cache for weak networks, command reply flow, and exponential back‑off on reconnection.

7. Server‑side ingestion (Spring Boot example)

package com.example.iot.telemetry;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Slf4j
@Service
@RequiredArgsConstructor
public class TelemetryIngestionService {
    private final ObjectMapper objectMapper;
    private final StringRedisTemplate redisTemplate;
    private final DeviceEventRepository deviceEventRepository;
    private final DeviceShadowRepository deviceShadowRepository;
    private final DeviceAlertService deviceAlertService;

    @KafkaListener(topics = "${iot.kafka.telemetry-topic}", groupId = "${spring.application.name}", concurrency = "${iot.kafka.telemetry-consumers:6}")
    public void onMessage(String raw) {
        try {
            TelemetryMessage message = objectMapper.readValue(raw, TelemetryMessage.class);
            ingest(message);
        } catch (Exception ex) {
            log.error("telemetry message parse failed, payload={}", raw, ex);
            throw new IllegalStateException("telemetry parse failed", ex);
        }
    }

    @Transactional
    public void ingest(TelemetryMessage message) {
        String idempotentKey = "iot:telemetry:dedup:" + message.eventId();
        Boolean firstSeen = redisTemplate.opsForValue()
                .setIfAbsent(idempotentKey, "1", Duration.ofHours(6));
        if (Boolean.FALSE.equals(firstSeen)) {
            log.info("duplicated telemetry ignored, eventId={}", message.eventId());
            return;
        }
        try {
            deviceEventRepository.insert(DeviceEventEntity.from(message));
        } catch (DuplicateKeyException ex) {
            log.info("duplicated telemetry ignored by db unique constraint, eventId={}", message.eventId());
            return;
        }
        DeviceShadowEntity current = deviceShadowRepository
                .findByTenantAndDeviceId(message.tenant(), message.deviceId())
                .orElse(DeviceShadowEntity.init(message));
        current.apply(message.metrics(), Instant.ofEpochMilli(message.timestamp()));
        current.setOnline(true);
        current.setLastSeenAt(Instant.ofEpochMilli(message.timestamp()));
        deviceShadowRepository.save(current);
        deviceAlertService.evaluate(message, current);
    }

    @JsonIgnoreProperties(ignoreUnknown = true)
    public record TelemetryMessage(
            String eventId,
            String tenant,
            String productKey,
            String deviceId,
            long timestamp,
            Map<String, Double> metrics) {}
}

This service shows the required idempotency (Redis + DB unique key), state convergence into device_shadow, and the separation of raw event storage from business state.

8. Command flow – outbox pattern and state machine

Directly publishing a command after writing to the command table creates two failure windows (DB commit vs publish, publish vs state update). The robust approach uses a local transaction that writes both the command record (PENDING) and an outbox event, then an asynchronous publisher reads pending outbox rows, publishes to MQTT, marks the outbox as SENT, and updates the command status.

package com.example.iot.command;

import java.time.Instant;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
@RequiredArgsConstructor
public class CommandApplicationService {
    private final DeviceCommandRepository deviceCommandRepository;
    private final CommandOutboxRepository commandOutboxRepository;

    @Transactional
    public CreateCommandResponse createCommand(CreateCommandRequest request) {
        String commandId = UUID.randomUUID().toString();
        DeviceCommandEntity command = DeviceCommandEntity.pending(
                commandId, request.tenant(), request.productKey(), request.deviceId(),
                request.commandName(), request.payload(), Instant.now(),
                Instant.now().plusSeconds(request.timeoutSeconds()));
        deviceCommandRepository.save(command);
        CommandOutboxEntity outbox = CommandOutboxEntity.pending(
                UUID.randomUUID().toString(), commandId, command.buildTopic(),
                command.payloadJson(), Instant.now());
        commandOutboxRepository.save(outbox);
        return new CreateCommandResponse(commandId, "PENDING");
    }
}

The publisher component locks a batch of pending outbox rows, publishes each via MQTT, marks the outbox and command as SENT, and retries on failure with exponential back‑off.

9. Common pitfalls and engineering safeguards

Connection storms : batch device restarts or regional outages can cause a sudden surge of CONNECT packets. Mitigate with exponential back‑off on the device, LB rate‑limiting, and regional sharding.

Hot topics : avoid funneling all devices into a single MQTT or Kafka topic; use hierarchical topics (tenant/product/device) and hash‑based Kafka partitions on deviceId or tenant+deviceId.

Back‑pressure : never let the broker become an infinite buffer. Bridge to Kafka with flow control, and apply downstream throttling or graceful degradation (sampling, pausing non‑critical analytics).

Idempotency : treat MQTT QoS 1 as “at least once”. Use a business‑level deduplication key (e.g., eventId) stored in Redis (short‑term) and a DB unique constraint (long‑term).

Batching : write telemetry, alerts, and shadow updates in bulk to TSDB, Redis, or relational stores to reduce per‑message overhead.

Broker is not a database : never rely on MQTT for durable command history, audit, or replay; funnel those to Kafka or a dedicated event store.

QoS ≠ business consistency : still need transactions, outbox, and state machines for exactly‑once semantics.

Will messages : empty payloads may delete retained messages in many implementations – avoid accidental state loss.

Shared subscriptions : useful for scaling consumers but do not replace a proper event bus; they still need idempotency and replay handling.

Device time drift : record both device‑generated timestamp and platform ingest timestamp; separate event_time from ingest_time in downstream analytics.

10. Deployment roadmap

Stage 1 : single‑node MQTT broker for PoC; validate topic design and device SDK.

Stage 2 : broker cluster + Kafka bridge; introduce rule engine, device shadow service, command state service, and observability.

Stage 3 : multi‑region edge brokers; regional load balancing, cross‑region routing, and central event aggregation.

11. Monitoring & alerting

Access layer metrics (connections, new‑connection rate, disconnect rate, inbound/outbound TPS, inflight count, rule‑engine failures) answer "is the gateway healthy?".

Business layer metrics (upstream latency, Kafka lag, shadow refresh latency, pending command count, ACK timeout rate, device online rate, alert rule latency) answer "are messages converging to state in time?".

Key tracing IDs – eventId, commandId, deviceId, tenant – must be propagated through broker logs, Kafka logs, shadow updates, and command processing to enable end‑to‑end debugging.

Top alerts :

Connection‑storm detection (sudden spike in new connections).

Command ACK timeout rate exceeding threshold.

Tenant or region online‑rate drop.

12. Capacity planning

Estimate four capacities: connections, upstream telemetry, downstream command peaks, and state‑store writes. Example: 200 k devices, baseline telemetry every 10 s → 20 k msg/s; 20 % of devices in peak reporting every 2 s adds another 20 k msg/s, yielding ~40 k msg/s peak. Add command bursts (e.g., 3 k cmd/s) and auxiliary events to size broker, Kafka, and storage layers.

Load‑test not only the broker but also the full pipeline: connection storms, sustained telemetry, batch command bursts, ACK bursts, Kafka consumption, and shadow writes.

13. Production checklist

Device authentication (TLS, certificates, ClientId format).

Will message configuration.

Per‑tenant connection quotas and rate limits.

MQTT‑to‑Kafka bridge with proper partition key.

Device shadow schema and write path.

Command state machine (PENDING, SENT, ACKED, FAILED, TIMEOUT, CANCELLED).

Outbox persistence and retry logic.

Governance ACLs, throttling, gray‑release, certificate rotation.

Device‑side reconnection back‑off, offline cache, command reply handling.

Time‑sync strategy (store device time and ingest time separately).

14. When MQTT may be overkill

If the device count is tiny, latency requirements are lax, the primary client is a browser, or the workload is simple request‑response RPC, alternatives such as SSE, WebSocket, or plain HTTP may be sufficient.

15. Final take‑away

Moving from HTTP polling to MQTT is not merely a protocol swap; it forces a shift from short‑lived request semantics to a long‑lived session, event‑driven processing, and a multi‑plane architecture that can handle peaks, back‑pressure, and observability. Only when the access, event, state, and governance planes are all built does the system achieve true production‑grade high‑concurrency reliability.