Handling Multiple MQTT Client Connections Using Python

It is common to require several connections either to a single broker or to multiple brokers.

There a several ways of achieving this and we will examine two ways.

First let us recap on what we require with a single connection.

• unique client_id
• client loop to process callbacks
• broker address and port.

If we have multiple client connections then we normally need a:

• unique client_id for each connection
• A client loop for each connection
• Broker address and port details for each connection.

Multiple Clients and Threads loop() and loop_start()

On a Single client scenario we can send and receive using a single thread if we use the client.loop() function instead of the loop_start and loop_stop functions.

However this isn’t normally done as it is easier to use the loop_start() and Loop_stop() as they handle reconnects automatically, whereas the loop() function doesn’t and so you need to do that yourself.

If we use the loop_start() function for a single client then we have two threads.

Having two threads as opposed to one isn’t a problem but if you want to implement several thousand clients then it is.

However for 10 or twenty clients then it is probably OK

Threading Advantages

The main advantage of using threads is speed. If your script needs to be very responsive then your will probably need threads.

Tutorial Overview

I have tried various methods of implementing multiple clients and they vary in complexity.

The main problem with multiple clients is:

• Handling of the client Connections and reconnects
• Handling of the callbacks

We will cover:

• Multiple clients Using Loop_start() and loop_stop() functions.
• Multiple clients Using Loop() function
• Multiple Clients using Threads.

Method 1 – Multiple Client Loops.-Loop_start()

For multiple client connections then each client will need a loop.

Therefore if you have 2 client connections you will need two loops which equals two additional threads.

For 20 client connections you will need twenty loops which equals twenty additional threads.

The code below show how we create unique client_ids and multiple loops for multiple clients,

import paho.mqtt.client as mqtt
clients=[]
nclients=20
mqtt.Client.connected_flag=False
#create clients
for i  in range(nclients):
   cname="Client"+str(i)
   client= mqtt.Client(cname)
   clients.append(client)
for client in clients:
  client.connect(broker)
  client.loop_start()

Code explanation
1. We use an array to store the client connections.
2. The first for loop create a unique client id and uses that id to create a new connection.
3. It then stores the client connection reference in an array.
4. The second for loop goes through the connection array and uses the client object to create the client loop.
5. We now have 20 loops active and 20 additional threads. (nclients=20).

How do we publish and/or subscribe?

We just get client object from the array

for client in clients: 
  client.connect(broker)
  client.subscribe(topic)
  client.publish(topic,message)

Alternatively we can subscribe in the on_connect callback.

def on_connect(client, userdata, flags, rc):

    if rc==0:
        client.connected_flag=True
        client.subscribe(topic)

Note: You don’t need separate callbacks for each client as the callback contains the client object

Receiving Messages
Uses the standard on_message callback

def on_message(client, userdata, message):
   print("message received",str(message.payload.decode("utf-8")))

Notes:
The approach works well if all of the clients connect to the same broker, and do the same thing.

It is easy to modify the script to publish and subscribe to different auto generated topics using the same technique as used in creating the unique client id.

However using different brokers isn’t so simple.

Uses: I’ve used this style for creating stress test scripts where you need lots of clients that publish and subscribe.

Demo script: The demo script demo1.py connects and publishes and subscribes on various topics.

Received messages are displayed along with the thread count.

Example :

 Modification for Multiple Brokers, and Topics

In this alternative we use a Python dictionary to store the client details connection,subscribe and publish details which gives use more flexibility.

However the price we pay is increased initial configuration.

Code:

client1={broker1:192.168.1.159,port:1883,name:”blank”,sub_topic:”test1″,pub_topic:”test1″,client=””}

For multiple clients we add them to an array or dictionary.

clients=[
{"broker":"192.168.1.159","port":1883,"name":"blank","sub_topic":"test1","pub_topic":"test1"},
{"broker":"192.168.1.65","port":1883,"name":"blank","sub_topic":"test2","pub_topic":"test2"}
]

Notes: I’ve left the client_id as blank as it is better to create a random name using techniques like this:

t=int(time.time()) #remove decimal part
t=str(t) #concert to string
client_id=client1+t
and
client_id=client2+t
etc

we add it to the dictionary using:

clients[0]["name"]=client_id

You could also add the client object using:

client=mqtt.Client(client_id)
clients[0]["client"]=client

In that way you have references between the client object,broker,topics etc.

Demo script: The demo script demo1a.py uses an array and dictionary to store client data and is suitable when dealing with multiple brokers and topics. Otherwise it does the same as demo1.py

Method 2 – Using The loop() function

If we use the loop() function we can reduce the number of threads.

The demo script demo1b.py is the same as demo1a.py but uses loop() and not loop_start().

You should also note and test that the main while loop is a bad example when using the loop() call as the while loop has lots or sleep calls.

To counter this it is better to use a separate thread to call the while loop.

Here is the code:

def multi_loop(nclients):
   global run_flag 
   while run_flag: 
      for i in range(nclients):
         client=clients[i]["client"]
         client.loop(0.01)

To start in a new thread use:

t = threading.Thread(target=multi_loop,args=(nclients,True))#start multi loop
t.start()

Method 3 – Using Threads for Each Client

If you need each client to be very responsive then you will need to start a thread for each client

The script demo2.py uses a different thread for each client connection.

The thread calls the client loop() function manually and there is no need for loop_start().

The script handles reconnects and will attempt 3 before stopping that client thread.

Total threads =number of clients+1 .

Note: If started using IDE then add another thread.

The clients do nothing at present to make them do something edit the pub function.

You can create a different pub function for each client if needed but you will need to modify how the threads are created if you do.

Course Links

1. Introduction to the Paho Python MQTT Client
2. Introduction to the Client Class
3. Connecting to a Broker
4. Publishing Using The Paho Python MQTT Client
5. –Subscribing using The Paho Python Client
6.  Receiving Messages with the Paho MQTT Python Client
7. Understanding The Loop
8. Understanding Callbacks
9. Handling Multiple Client Connections