sanic+mqtt使用

最近项目需要，在sanic中搭配使用mqtt，由于sanic是一个协诚异步框架，所以也必须使用mqtt的异步版本，常见的有：hbmqtt (项目已经不更新，版本太低不匹配)，asyncio-mqtt，asyncio-paho。这里我们使用asyncio-paho库。

代码如下：

import json
from sanic import Sanic, response
from asyncio_paho import AsyncioPahoClient

app = Sanic(__name__)

# mqtt配置
host = '192.168.0.15'
port = 1883
topic = 'mytopic'
client_id = 'fsdfdsdsfs'


@app.listener('before_server_start')
async def server_init(app, loop):
    app.client = AsyncioPahoClient(client_id, loop=loop)
    app.client.asyncio_listeners.add_on_connect(on_connect_async)
    app.client.asyncio_listeners.add_on_message(on_message_async)
    await app.client.asyncio_connect(host=host, port=port)


async def on_connect_async(client, userdata, flags_dict, result):
    await client.asyncio_subscribe(topic=topic)


async def on_message_async(client, userdata, msg):
    """
    消息处理函数
    """
    print('====', json.loads(msg.payload))


@app.route('/f1', methods=['post'])
async def f1(request):
    d = request.json
    request.app.client.publish(
        topic,
        payload=json.dumps(d)
    )
    return response.text('success')


if __name__ == '__main__':
    app.run('0.0.0.0', port=3331)

可以看到，代码比较简单，核心在于服务启动之前，初始化好mqtt客户端即可。

启动服务，调用http://localhost:3331/f1 。控制台输出订阅的消息

[2022-07-13 15:20:03 +0800] - (sanic.access)[INFO][127.0.0.1:52116]: POST http://localhost:3331/f1  200 7
==== {'key': '1234567'}

成功~