本文是作者之前做的一个开源项目，用Python及MQTT写的一个物联网网关，源码托管在github上。之前在国外社区发了一个贴，现在转到博客园来。
We always face with some applications which needs to just send some data to a server. To start thinking about MQTT-based framework, here’s the application scenarios:
What’s MQTT
MQTT is a machine-to-machine (M2M)/”Internet of Things” connectivity protocol. It was designed as an extremely lightweight publish/subscribe messaging transport. It is useful for connections with remote locations where a small code footprint is required and/or network bandwidth is at a premium. For example, it has been used in sensors communicating to a broker via satellite link, over occasional dial-up connections with healthcare providers, and in a range of home automation and small device scenarios.
Pub/Sub Model
Figure: Pub/Sub pattern
In a pub/sub model, the producer acts as publisher, and consumer acts as subscriber. Using the publish – subscribe pattern in application can dramatically simplify its design and improve testability.
In a distribute IoT system, basically you just have some part(s) of your device subscribe to a particular topic and have some other part(s) of your device publish messages with that topic. Messages sent will be delivered to all registered handlders (aka listeners, also receivers) for a given topic(one-to-many broadcast).
We implement the following parts:
1.Node
1.1 Generate a heartbeat in zigbee Node
every node can produce data, adc value, temperature etc.
void generateHeartbeat() { heartbeat_t heartbeat; sendtoAir(heartbeat); }
The heartbeat format gateway requires is key-value pair string (KVP).
“key:value” + ‘ ’
eg: “temperature:23 ”
Heartbeat is sent with no ack.
Figure: Network Topology
2.Gateway
Figure: Gateway software architecture
Gateway is written by pure Python, which makes it more pithy to implement the software components.
It consists of four parts:
1. Exchange component is one of the core components of the gateway project. It acts as a protocol converter between MQTT and PAN;
2. Data filter component provides some static filter, such as LinearFilter, EnumFilter etc. Once a filter is activated by user, each incoming/outgoing data frame will be filtered by it.
3. Plugin usually contains some user-defined business logic, such as sending a heartbeat, processing sensor data etc. Because every single data frame will pass to it, user can do anything they want according to the incoming data. Click here to get a depth understanding of the plugin developing.
4. PAN device Driver is a layer used for communication between Pan Device and Host CPU. Usually, the physical connection is a serial port. Now, it supports MeshBee v1.0.4 firmware and XBee.
Main features:
1. Can be ported to any platform that supports Python.
2. User-defined Qos, which ensure the reliability of data transmission.
3. Message forwarding is handled by Broker. User can focus on their application developing.
4. Pub/Sub model, one-to-many message deliver.
5. light-weight messaging protocol over TCP. It is designed for connections with remote locations where a “small code footprint” is required and/or network bandwidth is limited.
6. Support plugins expanding.
7. Support data filter.
8. Support any device that has a serial port.
9. Build-in Sqlite Database.
Getaway running flow
The following flow provides some basic procedures of how a gateway engine run.
Gateway configuration
Gateway program will read its configuration files at startup. We create a yaml file in root directory. Modify the following items to meet your request.
1. yaml['pan']['port']
Pan device serial port device ID in /dev.
2. yaml['pan']['baudrate’']
Baudrate of pan device serial port device.
3. yaml['pan']['driver_class']
Specified pan device driver
4. yaml['mqtt’']['client_id']
Gateway client ID
5. yaml['mqtt’']['host']
Broker Url
6. yaml['mqtt’']['port']
Broker TCP port number.
3.Android App
An android app can control and monitor every node from the pan network. It runs a background service to server MQTT.
private void StartClientThread(){ tpMap = new HashMap(); tpMap.put("topic_pool", CONST_TOPIC); //will be execute only once //service in android is running on main thread,so creating a new //thread is the best way to avoid blocking new Thread(new Runnable() { @Override public void run() { //background task try{ mqttClient = new MqttClient(BROKER_URL, clientId, new MemoryPersistence()); mqttClient.setCallback(new MqttCallback(){ @Override public void messageArrived(MqttTopic topic, MqttMessage message) throws Exception{ String strTopic = topic.getName(); //if here comes the constant topic message if(strTopic.equals(tpMap.get("topic_pool").toString())){ //parse it, format: topic1@topic2@topic3 String raw_topic = new String(message.getPayload()); String[] arrTopic = raw_topic.split("@"); for(int i = 0; i < arrTopic.length; i++){ if(arrTopic[i].indexOf("dio9") > 0){ tpMap.put("dio9", arrTopic[i]); } else if(arrTopic[i].indexOf("a3") > 0){ tpMap.put("a3", arrTopic[i]); } mqttClient.subscribe(arrTopic[i]); } Alert("gateway is starting"); } else{ //other topic MqttMessageHandler(strTopic, message); } } @Override public void connectionLost(Throwable cause){ } @Override public void deliveryComplete(MqttDeliveryToken token){ } }); //connect to broker and subscribe topic_pool mqttClient.connect(); //blocking method mqttClient.subscribe(tpMap.get("topic_pool").toString()); mqttClient.subscribe(OfflineAlertTopic); //query topic list from gateway MqttTopic mqttTopic = mqttClient.getTopic(GwListeningTopic); final MqttMessage mqttMessage = new MqttMessage(tpMap.get("topic_pool").toString().getBytes()); mqttTopic.publish(mqttMessage); } catch(Exception e){ Toast.makeText(getApplicationContext(), "Something went wrong!" + e.getMessage(), Toast.LENGTH_LONG).show(); e.printStackTrace(); } } }).start(); }