In-Depth Analysis of Smart-MQTT Clustering Technology Based on the Feat Framework

Building an Efficient MQTT Cluster System 🌐

Introduction 🧭

In modern distributed systems, the clustered deployment of MQTT Brokers is a key method to enhance system availability βœ…, scalability πŸ”‹, and load balancing capabilities πŸ”„.

The <span>cluster-plugin</span> of smart-mqtt implements an efficient and scalable cluster coordination system 🌟 by combining the HTTP client of the Feat framework with the Server-Sent Events (SSE) mechanism. This plugin supports deployment modes for Core nodes and Worker nodes, enabling node discovery, state synchronization, and message forwarding πŸ“‘ through HTTP communication.

This article will delve into its technical details based on the source code implementation, including node coordination mechanisms, message distribution and reception processes, queue strategy handling, etc. πŸ’‘, helping developers better understand its design principles and deployment methods.

1. Overall Architecture Overview 🧱

1.1 Core Component:<span>Coordinator</span>

The core component of the plugin is the <span>Coordinator</span> class, which inherits from <span>AsyncTask</span> and is responsible for starting two key threads:

  • β€’ Distributor (Distribution Thread): Responsible for consuming messages from the message bus (<span>MessageBus</span>) and distributing them to other nodes in the cluster πŸ“€. This is specifically implemented within <span>Distributor</span>.
  • β€’ Receiver (Receiving Thread): Responsible for receiving messages from other nodes and publishing them to the local message bus πŸ“₯. This is specifically implemented within <span>Receiver</span>.

1.2 Configuration Management:<span>PluginConfig</span>

Configuration is loaded through <span>PluginConfig</span>, which includes the following key parameters:

  • β€’ Whether it is a Core node (Core) πŸ“
  • β€’ Listening address and port πŸ“
  • β€’ Queue length and strategy βš™οΈ
  • β€’ List of cluster node addresses πŸ“‹

1.3 Startup Process 🚦

During plugin initialization, <span>ClusterPlugin</span> will start the <span>Coordinator</span> thread. If the node is Core, it will start the HTTP server provided by Feat, listening for requests on the <span>/cluster</span> path to handle status checks, message pushing, and subscription operations πŸ“‘.

// ... existing code ...
if (pluginConfig.isCore()) {
    httpServer = FeatCloud.cloudServer(cloudOptions -> cloudOptions.registerBean("mqttSession", coordinator.mqttSession).registerBean("brokerContext", brokerContext).host(pluginConfig.getHost()).port(pluginConfig.getPort()).debug(true)).listen();
}
// ... existing code ...

2. Node Coordination and Discovery Mechanism 🀝

<span>Coordinator</span> maintains a list of <span>ClusterClient</span>, each corresponding to a cluster node (loaded via <span>pluginConfig.getClusters()</span>).

2.1 Node Status Detection πŸ•΅οΈβ™‚οΈ

Each <span>ClusterClient</span> periodically sends GET requests to the <span>/cluster/status</span> interface using Feat’s <span>HttpClient</span> to check if the node is available:

// ... existing code ...
clusterClient.httpClient.get("/cluster/status").onSuccess(httpResponse -> {
    LOGGER.info("check node status success.");
    clusterClient.httpEnable = true;
    clusterClient.checkPending = false;
}).onFailure(throwable -> {
    clusterClient.httpEnable = false;
    clusterClient.checkPending = false;
    LOGGER.error("check node status error", throwable);
}).submit();
// ... existing code ...

βœ… Success: Marked as available, start SSE subscription❌ Failure: Marked as unavailable, pause connection

2.2 SSE Subscription Mechanism πŸ“»

  • β€’ Core Node: Starts SSE subscriptions for each available ClusterClient to receive messages from other nodes πŸ“₯.
  • β€’ Worker Node: Connects to only one Core node and automatically releases and reallocates when its connection fails πŸ”.

Subscriptions are initiated via Feat’s <span>HttpClient</span> by sending a POST request to <span>/cluster/subscribe/{nodeType}/{access_token}</span> and using a custom <span>ClusterMessageStream</span> to handle SSE events.

3. Message Distribution Mechanism (Distributor) πŸ“€

The Distributor thread consumes messages from the <span>MessageBus</span> and distributes them based on node type:

  • β€’ Core Node: Sends messages via HTTP POST to the <span>/cluster/put/core</span> interface of other Core nodes and publishes the event <span>CLIENT_DIRECT_TO_CORE_BROKER</span> to notify local Workers πŸ“’.
  • β€’ Worker Node: Sends messages to the <span>/cluster/put/worker</span> interface of its connected Core node πŸ“€.
// ... existing code ...
if (pluginConfig.isCore()) {
    for (ClusterClient clusterClient : clients) {
        if (clusterClient.httpEnable) {
            // coreθŠ‚η‚Ήεˆ†ε‘ζΆˆζ―θ‡³ι›†ηΎ€ε…Άδ»–coreθŠ‚η‚Ή
            clusterClient.httpClient.post("/cluster/put/core").header(header -> header.keepalive(true).set("access_token", ACCESS_TOKEN).setContentLength(message.getPayload().length).set(ClusterController.HEADER_TOPIC, message.getTopic().getTopic())).body(requestBody -> requestBody.write(message.getPayload())).onFailure(throwable -> {
                clusterClient.httpEnable = false;
                LOGGER.error("send message to cluster error", throwable);
            }).onSuccess(httpResponse -> {
                LOGGER.info("send message to cluster success");
            }).submit();
        }
    }
    brokerContext.getEventBus().publish(ClusterPlugin.CLIENT_DIRECT_TO_CORE_BROKER, message);
} elseif (workerClient != null) {
    workerClient.httpClient.post("/cluster/put/worker").header(header -> header.keepalive(true).set("access_token", ACCESS_TOKEN).setContentLength(message.getPayload().length).set(ClusterController.HEADER_TOPIC, message.getTopic().getTopic())).body(requestBody -> requestBody.write(message.getPayload())).onFailure(throwable -> {
        workerClient.httpEnable = false;
        LOGGER.error("send message to cluster error", throwable);
    }).submit();
}
// ... existing code ...

3.1 Queue Strategy πŸ“¦

Messages are first placed into the <span>distributorQueue</span> (ArrayBlockingQueue), with its capacity determined by <span>queueLength</span>. If the queue is full:

  • β€’ Strategy 0: Discard the latest message ❌ (default)
  • β€’ Strategy 1: Discard the oldest message ❌
private void offer(Message message, ArrayBlockingQueue<Message> clusterMessageQueue) {
    if (pluginConfig.getQueuePolicy() == QUEUE_POLICY_DISCARD_NEWEST) {
        boolean suc = clusterMessageQueue.offer(message);
        if (!suc) {
            LOGGER.warn("queue is full, discard message: {}", message);
        }
    } else {
        while (!clusterMessageQueue.offer(message)) {
            Message discard = clusterMessageQueue.poll();
            if (discard != null) {
                LOGGER.warn("queue is full, discard message: {}", discard);
            }
        }
    }
}

4. Message Reception Mechanism (Receiver) πŸ“₯

The Receiver thread consumes messages from the <span>receiverQueue</span> and publishes them to the local <span>MessageBus</span>.

4.1 SSE Message Subscription πŸ“»

By sending a request to <span>/cluster/subscribe/{nodeType}/{access_token}</span>, subscriptions to SSE message streams are initiated. The custom <span>ClusterMessageStream</span> parses the <span>topic</span>, <span>payload</span>, and <span>retained</span> flags in the event and constructs a <span>Message</span> object:

@Override
public void onEvent(HttpResponse httpResponse, String topic, byte[] payload, boolean retained) {
    if (!enabled) {
        LOGGER.warn("cluster-plugin-consume-message-error");
        return;
    }
    Message message = new Message(brokerContext.getOrCreateTopic(topic), MqttQoS.AT_MOST_ONCE, payload, retained);
    offer(message, receiverQueue);
}

4.2 Message Broadcasting Mechanism πŸ“’

After receiving a message, the Core node further broadcasts it to all Worker nodes πŸ”Š, implemented through <span>brokerContext.getEventBus().publish</span>.

5. Message Format and Parsing πŸ“„

The plugin uses a custom message format for serialization and deserialization:

topic:<topic_name>
retain:
payload:<length> <payload_bytes>
  • β€’ <span>topic</span> line indicates the message topic 🧾
  • β€’ <span>retain</span> line indicates whether it is a retained message (optional) πŸ“Œ
  • β€’ <span>payload</span> line contains the message body length and content πŸ“¦
public byte[] toBytes(Message message) {
    ByteArrayOutputStream byteOutputStream = new ByteArrayOutputStream(message.getPayload().length + 32);
    try {
        byteOutputStream.write(ClusterMessageStream.TAG_TOPIC);
        byteOutputStream.write(':');
        byteOutputStream.write(message.getTopic().getTopic().getBytes());
        byteOutputStream.write('\n');

        if (message.isRetained()) {
            byteOutputStream.write(ClusterMessageStream.TAG_RETAIN);
            byteOutputStream.write(':');
            byteOutputStream.write('\n');
        }
        byteOutputStream.write(ClusterMessageStream.TAG_PAYLOAD);
        byteOutputStream.write(':');
        byteOutputStream.write((message.getPayload().length + " ").getBytes());
        byteOutputStream.write(message.getPayload());
        byteOutputStream.write('\n');
        byteOutputStream.write('\n');
    } catch (IOException e) {
        e.printStackTrace();
    }
    return byteOutputStream.toByteArray();
}

6. Configuration and Deployment Recommendations βš™οΈ

6.1 Plugin Configuration (<span>plugin.yaml</span>)

core: true                # Whether it is a core node, true=core node, false=worker node
host: 0.0.0.0             # Cluster service listening address, effective only when core is true
port: 8884                # Cluster service listening port, effective only when core is true
queueLength: 1024         # Message queue length
queuePolicy: 0            # Queue strategy (0=discard newest, 1=discard oldest)
clusters:                 # List of cluster node addresses
- http://core1:8884
- http://core2:8884

6.2 Deployment Considerations πŸ“Œ

  • β€’ Startup Order: Start Core nodes first, then Worker nodes, to ensure node discovery and connection are normal πŸš€.
  • β€’ Network Connectivity: Ensure that nodes can communicate through the configured HTTP ports 🌐.
  • β€’ Queue Tuning: Adjust <span>queueLength</span> and <span>queuePolicy</span> based on business load to avoid message loss or backlog βš–οΈ.

7. Conclusion 🧾

<span>cluster-plugin</span> implements an efficient and flexible MQTT cluster architecture 🌟 through the HTTP client and SSE mechanism provided by the Feat framework. Its core design includes:

  • β€’ Asynchronous task scheduling (Distributor/Receiver) πŸ”„
  • β€’ Queue management and strategy control βš™οΈ
  • β€’ Event-driven message distribution and reception πŸ“’
  • β€’ Dynamic node discovery and state synchronization mechanism πŸ”

This plugin is suitable for high availability and high concurrency MQTT cluster scenarios πŸ’Ό. Developers can adjust configuration parameters and optimize queue strategies according to actual needs to further enhance system performance and stability πŸ“ˆ.

Gitee Address：smart-mqtt/plugins/cluster-plugin[1]Documentation Reference：Feat Framework Documentation[2]

If further customization or integration is needed, feel free to participate in open-source community discussions and contributions πŸ’¬βœ¨

Reference Links

<span>[1]</span> smart-mqtt/plugins/cluster-plugin: https://gitee.com/smart-mqtt/smart-mqtt<span>[2]</span> Feat Framework Documentation: https://smartboot.tech/feat

Leave a Comment