Client connection to VerneMQ cluster

Question

We set up a VerneMQ 3 nodes cluster (https://vernemq.com/docs/clustering/).

The problem is I didn’t find any example on how to connect the clients to the whole cluster.

E.g. for a single VerneMQ instance the client connection (in python) is:

client.connect("xxx.xxx.xxx.xxx", 1883, 60)

where "xxx.xxx.xxx.xxx" is the address of the single instance.
In case of cluster is there a way to specify the address of the whole cluster, instead of pointing a single node?

Answer 1

There is no such thing as the "specific address of the whole cluster". (for VerneMQ and for any other clustered server software).

If you need a single point of contact, you need a load balancer in front of the cluster. You can then use different load balancing strategies.

VerneMQ works well with HAProxy for instance and supports the proxy protocol.

Hope this helps.

Answer 2

With my client written using Paho MQTT client library, I ended up in using the following extension of Client class:

import random
from paho.mqtt.client import *


class ClusterClient(Client):
    """A subclass of paho.mqtt.Client that supports connecting to a cluster of
       mqtt brokers. connect() and connect_async() additionally accept a list
       of hostnames or host/port tuples:

           connect("host1")

           connect(["host1", "host2", "host3"]) # use default port 1883

           connect(["host1", ("host2", 8883), ("host3", 8883)])

       Hosts to connect to are chosen randomly. If a host disappears the client
       automatically connects to another host from the list.
    """

    def __init__(self, client_id="", clean_session=True, userdata=None,
            protocol=MQTTv311, transport="tcp"):
        super().__init__(client_id, clean_session, userdata, protocol, transport)
        self._hosts = []

    def connect_async(self, host, port=1883, keepalive=60, bind_address=""):
        if isinstance(host, (list, tuple)):
            self._hosts = [(t, 1883) if isinstance(t, str) else t for t in host]
        else:
            self._hosts = [(host, port)]

        for host, port in self._hosts:
            if host is None or len(host) == 0:
                raise ValueError('Invalid host.')
            if port <= 0:
                raise ValueError('Invalid port number.')

        host, port = random.choice(self._hosts)

        super().connect_async(host, port, keepalive, bind_address)

    def reconnect(self):
        hosts = self._hosts[:]
        random.shuffle(hosts)
        while True:
            self._host, self._port = hosts.pop(0)
            try:
                return super().reconnect()
            except socket.error:
                if not hosts:
                    raise