Production‑Grade IM Architecture with MQTT over RabbitMQ: Principles & Practices

Why MQTT over RabbitMQ instead of a custom WebSocket gateway

Typical first‑generation IM systems embed a self‑written WebSocket server that handles connection management, routing, offline storage, QoS, and keep‑alive in the application layer. This works for a few thousand connections but collapses at tens of thousands because connection handling, message routing, session consistency, offline compensation, back‑pressure and operability become bottlenecks.

Applicable scenarios and limits

Suitable for

Enterprise IM, customer‑service, notification push, lightweight social messaging

Unified access for apps, desktop clients and IoT devices

Systems already using RabbitMQ extensively

Fine‑grained topic routing and strong integration between MQTT and AMQP services

High operational consistency, observability and scalability requirements

Not suitable for

Pure browser chatrooms that only need simple WebSocket push

Very large pure IoT platforms that need a dedicated MQTT ecosystem

Use cases that require deep custom MQTT broker optimisations

Extreme throughput or fan‑out where the team lacks RabbitMQ expertise

Core principle

Embed MQTT capabilities into an existing micro‑service ecosystem: MQTT solves massive connection handling, while RabbitMQ provides enterprise‑grade routing, back‑pressure, asynchronous decoupling and operational tooling.

Protocol mapping (MQTT ↔ RabbitMQ/AMQP)

Client : TCP connection + RabbitMQ connection

Client ID : connection identity / session identifier

Topic : topic exchange routing key

Publish : basic.publish Subscribe : queue + binding + consumer

Session : subscription state + unacknowledged message state

QoS 0/1 : different acknowledgement strategies and delivery semantics

Retained message : broker‑kept latest message per topic

Will message : broker‑side message triggered on abnormal disconnect

Message flow example

A client publishes to im.upstream.chat. The bridge service consumes the message, performs authentication, rate‑limiting, idempotency checks, persists an inbound index, queries the receiver’s presence, and then either publishes to im.user.{receiverId}.msg for online users or writes the payload to a Redis Streams offline inbox for offline users.

Why split online and offline paths

Online path needs low latency and fast routing.

Offline path needs durability, compensation and historical queries.

Mixing both forces a single queue to handle both low‑latency traffic and large backlogs, and makes consumers handle both real‑time push and heavy storage, causing blocking when many users disconnect.

Recommended split:

RabbitMQ handles online routing and asynchronous decoupling.

Redis Streams (or a database) store offline inboxes and provide historical queries.

Business services decide online/offline strategy per message.

End‑to‑end processing pipeline

Client publishes to im.upstream.chat Bridge service consumes, authenticates, validates and deduplicates

Persist inbound index

Check receiver presence

If online, publish to im.user.{uid}.msg If offline, append to

stream:offline:{uid}

Topic design principles

A good topic must be readable, stable for routing, have clear permission boundaries, be easy to monitor and avoid unbounded cardinality.

Recommended pattern: im.{domain}.{scope}.{id}.{action} Examples:

im.user.1024.msg
im.user.1024.ack
im.group.9001.msg
im.group.9001.notice
im.presence.1024.online
im.system.broadcast.notice

Avoid embedding high‑frequency dynamic fields or creating a new topic per message type because it leads to subscription explosion, monitoring difficulty and permission complexity.

Message envelope (standard JSON wrapper)

{
  "messageId": "01HV9T5JH6D3T3QFXGX0M5E4S2",
  "traceId": "8f3f4e921d3749f8898f1f88db4b1e2c",
  "bizType": "chat.text",
  "senderId": 1001,
  "receiverId": 2001,
  "conversationId": "c2c_1001_2001",
  "clientSendTime": 1715600000123,
  "serverReceiveTime": 1715600000456,
  "qos": 1,
  "payload": {"text": "hello"},
  "ext": {"appVersion": "6.3.1", "deviceId": "iphone-15-1001"}
}

The envelope guarantees:

Idempotency via messageId Observability via traceId Protocol evolution via bizType Timeline analysis via separate client/server timestamps

Ordering guarantees

One‑to‑one conversations are partitioned by conversationId Group messages are partitioned by groupId Within a partition a single consumer thread preserves local order

Cross‑partition global order is not promised

Production‑grade Go MQTT client

The client adds automatic reconnection, resubscription, disables order enforcement (so a slow handler does not block the pipeline) and publish timeout control.

package mqttclient

import (
    "crypto/tls"
    "errors"
    "fmt"
    "log"
    "sync"
    "sync/atomic"
    "time"

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

type Config struct {
    Broker               string
    ClientID             string
    Username             string
    Password             string
    CleanSession         bool
    ConnectTimeout       time.Duration
    WriteTimeout         time.Duration
    KeepAlive            time.Duration
    PingTimeout          time.Duration
    MaxReconnectBackoff  time.Duration
    TLSConfig            *tls.Config
}

type Handler func(topic string, payload []byte)

type Client struct {
    cfg       Config
    client    mqtt.Client
    handlers  sync.Map
    connected atomic.Bool
    closed    atomic.Bool
}

func NewClient(cfg Config) *Client {
    if cfg.ConnectTimeout == 0 { cfg.ConnectTimeout = 5 * time.Second }
    if cfg.WriteTimeout == 0 { cfg.WriteTimeout = 3 * time.Second }
    if cfg.KeepAlive == 0 { cfg.KeepAlive = 30 * time.Second }
    if cfg.PingTimeout == 0 { cfg.PingTimeout = 10 * time.Second }
    if cfg.MaxReconnectBackoff == 0 { cfg.MaxReconnectBackoff = 1 * time.Minute }
    return &Client{cfg: cfg}
}

func (c *Client) Connect() error {
    opts := mqtt.NewClientOptions().
        AddBroker(c.cfg.Broker).
        SetClientID(c.cfg.ClientID).
        SetUsername(c.cfg.Username).
        SetPassword(c.cfg.Password).
        SetCleanSession(c.cfg.CleanSession).
        SetAutoReconnect(true).
        SetConnectRetry(true).
        SetConnectRetryInterval(3*time.Second).
        SetMaxReconnectInterval(c.cfg.MaxReconnectBackoff).
        SetKeepAlive(c.cfg.KeepAlive).
        SetPingTimeout(c.cfg.PingTimeout).
        SetConnectTimeout(c.cfg.ConnectTimeout).
        SetOrderMatters(false).
        SetTLSConfig(c.cfg.TLSConfig)

    opts.SetOnConnectHandler(func(cli mqtt.Client) {
        c.connected.Store(true)
        log.Printf("mqtt connected clientId=%s", c.cfg.ClientID)
        c.resubscribe(cli)
    })
    opts.SetConnectionLostHandler(func(cli mqtt.Client, err error) {
        c.connected.Store(false)
        log.Printf("mqtt connection lost clientId=%s err=%v", c.cfg.ClientID, err)
    })
    opts.SetReconnectingHandler(func(cli mqtt.Client, co *mqtt.ClientOptions) {
        log.Printf("mqtt reconnecting clientId=%s broker=%s", c.cfg.ClientID, c.cfg.Broker)
    })
    opts.SetDefaultPublishHandler(func(cli mqtt.Client, msg mqtt.Message) {
        if h, ok := c.handlers.Load(msg.Topic()); ok {
            h.(Handler)(msg.Topic(), append([]byte(nil), msg.Payload()...))
        }
    })

    c.client = mqtt.NewClient(opts)
    token := c.client.Connect()
    if !token.WaitTimeout(c.cfg.ConnectTimeout) {
        return errors.New("mqtt connect timeout")
    }
    if err := token.Error(); err != nil {
        return fmt.Errorf("mqtt connect failed: %w", err)
    }
    return nil
}

func (c *Client) Publish(topic string, qos byte, retained bool, payload []byte) error {
    if c.closed.Load() { return errors.New("mqtt client closed") }
    if !c.connected.Load() { return errors.New("mqtt client not connected") }
    token := c.client.Publish(topic, qos, retained, payload)
    if !token.WaitTimeout(c.cfg.WriteTimeout) {
        return fmt.Errorf("publish timeout topic=%s", topic)
    }
    return token.Error()
}

func (c *Client) Subscribe(topic string, qos byte, handler Handler) error {
    c.handlers.Store(topic, handler)
    token := c.client.Subscribe(topic, qos, func(cli mqtt.Client, msg mqtt.Message) {
        handler(msg.Topic(), append([]byte(nil), msg.Payload()...))
    })
    if !token.WaitTimeout(c.cfg.ConnectTimeout) {
        return fmt.Errorf("subscribe timeout topic=%s", topic)
    }
    return token.Error()
}

func (c *Client) resubscribe(cli mqtt.Client) {
    c.handlers.Range(func(key, value any) bool {
        topic := key.(string)
        handler := value.(Handler)
        token := cli.Subscribe(topic, 1, func(_ mqtt.Client, msg mqtt.Message) {
            handler(msg.Topic(), append([]byte(nil), msg.Payload()...))
        })
        if token.Wait() && token.Error() != nil {
            log.Printf("mqtt resubscribe failed topic=%s err=%v", topic, token.Error())
        }
        return true
    })
}

func (c *Client) Close() {
    if !c.closed.CompareAndSwap(false, true) { return }
    if c.client != nil && c.client.IsConnected() {
        c.client.Disconnect(250)
    }
}

Bridge service logic (Go)

package bridge

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    amqp "github.com/rabbitmq/amqp091-go"
)

type PresenceStore interface { IsOnline(ctx context.Context, userID int64) (bool, error) }

type OfflineStore interface { Append(ctx context.Context, userID int64, payload []byte) error }

type DownstreamPublisher interface { Publish(topic string, body []byte) error }

type MessageRepository interface { SaveInbound(ctx context.Context, msg ChatMessage) error }

type Deduplicator interface { TryMark(ctx context.Context, messageID string, ttl time.Duration) (bool, error) }

type ChatMessage struct {
    MessageID       string          `json:"messageId"`
    TraceID         string          `json:"traceId"`
    SenderID        int64           `json:"senderId"`
    ReceiverID      int64           `json:"receiverId"`
    ConversationID  string          `json:"conversationId"`
    BizType         string          `json:"bizType"`
    Payload         json.RawMessage `json:"payload"`
}

type Service struct {
    presence PresenceStore
    offline  OfflineStore
    pub      DownstreamPublisher
    repo     MessageRepository
    dedupe   Deduplicator
}

func (s *Service) HandleDelivery(ctx context.Context, d amqp.Delivery) error {
    var msg ChatMessage
    if err := json.Unmarshal(d.Body, &msg); err != nil {
        return fmt.Errorf("decode inbound message: %w", err)
    }
    ok, err := s.dedupe.TryMark(ctx, msg.MessageID, 24*time.Hour)
    if err != nil { return fmt.Errorf("dedupe check failed: %w", err) }
    if !ok { log.Printf("duplicated message ignored messageId=%s", msg.MessageID); return nil }
    if err := s.repo.SaveInbound(ctx, msg); err != nil { return fmt.Errorf("persist inbound message: %w", err) }
    online, err := s.presence.IsOnline(ctx, msg.ReceiverID)
    if err != nil { return fmt.Errorf("query presence failed: %w", err) }
    body, _ := json.Marshal(map[string]any{"messageId":msg.MessageID,"traceId":msg.TraceID,"conversationId":msg.ConversationID,"senderId":msg.SenderID,"payload":msg.Payload})
    if online {
        topic := fmt.Sprintf("im.user.%d.msg", msg.ReceiverID)
        return s.pub.Publish(topic, body)
    }
    return s.offline.Append(ctx, msg.ReceiverID, body)
}

AMQP consumer loop with back‑pressure and worker pool

func ConsumeForever(ctx context.Context, ch *amqp.Channel, svc *Service) error {
    if err := ch.Qos(500, 0, false); err != nil { return err }
    msgs, err := ch.Consume("im.upstream.chat.q", "", false, false, false, false, nil)
    if err != nil { return err }
    workerCount := 32
    sem := make(chan struct{}, workerCount)
    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        case msg, ok := <-msgs:
            if !ok { return amqp.ErrClosed }
            sem <- struct{}{}
            go func(d amqp.Delivery) {
                defer func(){ <-sem }()
                handleCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
                defer cancel()
                if err := svc.HandleDelivery(handleCtx, d); err != nil {
                    log.Printf("handle delivery failed messageId=%s err=%v", d.MessageId, err)
                    _ = d.Nack(false, true)
                    return
                }
                _ = d.Ack(false)
            }(msg)
        }
    }
}

Presence service (Redis)

type Status struct {
    Status       string `json:"status"`
    DeviceID     string `json:"deviceId"`
    ClientID     string `json:"clientId"`
    BrokerNode   string `json:"brokerNode"`
    LastActiveAt int64  `json:"lastActiveAt"`
}

type Store struct { rdb *redis.Client; ttl time.Duration }

func (s *Store) SetOnline(ctx context.Context, userID int64, st Status) error {
    key := fmt.Sprintf("presence:%d", userID)
    data, _ := json.Marshal(st)
    return s.rdb.Set(ctx, key, data, s.ttl).Err()
}

func (s *Store) IsOnline(ctx context.Context, userID int64) (bool, error) {
    key := fmt.Sprintf("presence:%d", userID)
    n, err := s.rdb.Exists(ctx, key).Result()
    return n == 1, err
}

Offline storage (Redis Streams)

type Store struct { rdb *redis.Client; maxLen int64; ttl time.Duration }

func (s *Store) Append(ctx context.Context, userID int64, payload []byte) error {
    key := fmt.Sprintf("stream:offline:%d", userID)
    pipe := s.rdb.TxPipeline()
    pipe.XAdd(ctx, &redis.XAddArgs{Stream: key, MaxLen: s.maxLen, Approx: true, Values: map[string]any{"payload": string(payload)}})
    pipe.Expire(ctx, key, s.ttl)
    _, err := pipe.Exec(ctx)
    return err
}

Reliability design (no loss, no duplication, local order)

No loss : client publishes with QoS 1, bridge service ACKs manually, inbound index persisted before forwarding, offline users get a Redis Stream entry, clients pull missed messages after reconnection.

No duplication : globally unique messageId, deduplication cache in the bridge, database unique constraints.

Local order : partition by conversationId or groupId, single‑threaded consumer per partition, optional worker‑affinity for ordering.

Scalability challenges and mitigations

Connection bottleneck : increase Erlang process limit (+P 1048576), raise OS file‑descriptor limit, use exponential back‑off on client reconnect, load‑balancer rate‑limiting.

Routing bottleneck : enforce strict topic template whitelist, avoid per‑device dynamic topics, limit prefetch/QoS (e.g., 200‑1000) to create back‑pressure.

Consumer bottleneck : worker pool with timeouts, circuit breakers for downstream services, rate‑limit high‑risk users/devices.

Horizontal expansion blueprint

RabbitMQ access layer – clustered horizontal scaling.

Bridge service – stateless horizontal scaling.

Presence store – Redis master‑slave or cluster.

Offline store – Redis cluster with sharded DB.

Message index – shard by session or user.

Kubernetes deployment

RabbitMQ runs as a StatefulSet to keep stable network IDs and persistent volumes. Bridge services are deployed as a stateless Deployment. Important considerations:

Four‑layer load‑balancing for MQTT.

Proxy protocol for real client IP.

Generous terminationGracePeriodSeconds for graceful connection teardown.

Readiness probes that wait for broker and consumer readiness.

Observability & troubleshooting

RabbitMQ metrics – connections, channels, queues, bindings, publish/confirm rates, ready/unacked counts, memory and disk watermarks, socket usage.

Bridge service metrics – inbound messages/sec, successful deliveries/sec, offline writes/sec, P95/P99 processing latency, ACK timeout count, retry count, idempotency hit count, downstream Redis/DB timeout count.

Critical log fields – traceId, messageId, conversationId, senderId, receiverId, topic, clientId, brokerNode.

Comparison with alternatives

Custom WebSocket gateway – provides lighter weight push but lacks enterprise‑grade protocol semantics, routing and back‑pressure.

Dedicated MQTT broker – offers deeper MQTT‑specific optimisations and massive device scale, but does not integrate natively with existing AMQP pipelines.

Roadmap to production

Phase 1 – Feasibility : single‑node RabbitMQ with MQTT plugin, basic client connect/publish/subscribe, unified message envelope, minimal bridge.

Phase 2 – Closed‑loop IM : implement presence, offline inbox, delivery ACK, idempotency, baseline monitoring.

Phase 3 – Production‑ready : add retry & dead‑letter queues, deploy on Kubernetes or stable cluster, conduct load‑testing and connection‑storm drills, address hot‑group and broadcast paths.

Phase 4 – Platformization : standardise topic conventions, ship SDKs, embed auth, rate‑limit, audit, expose a stable access protocol to downstream services.

Conclusion

MQTT solves connection management and messaging semantics; RabbitMQ solves routing, back‑pressure and operational governance. The decisive factor for a production‑grade IM system is whether the protocol capabilities can be turned into a full‑stack architecture that delivers reliability, scalability and observability.