怎么用Python开发EMQ X MQTT服务器插件
这篇文章主要讲解了"怎么用Python开发EMQ X MQTT服务器插件"，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习"怎么用Python开发EMQ X MQTT服务器插件"吧！
从 v4.1 版本开始，EMQ X MQTT 服务器 提供了专门的多语言支持插件 emqx_extension_hook ，现已支持使用其他编程语言来处理 EMQ X 中的钩子事件，开发者可以使用 Python 或者 Java 快速开发自己的插件，在官方功能的基础上进行扩展，满足自己的业务场景。例如：
验证某客户端的登录权限：客户端连接时触发对应函数，通过参数获取客户端信息后通过读取数据库、比对等操作判定是否有登录权限
记录客户端在线状态与上下线历史：客户端状态变动时触发对应函数，通过参数获取客户端信息，改写数据库中客户端在线状态
校验某客户端的 PUB/SUB 的操作权限：发布/订阅时触发对应函数，通过参数获取客户端信息与当前主题，判定客户端是否有对应的操作权限
处理会话 (Sessions) 和 消息 (Message) 事件，实现订阅关系与消息处理/存储：消息发布、状态变动时触发对应函数，获取当前客户端信息、消息状态与消息内容，转发到 Kafka 或数据库进行存储。
注：消息(Message) 类钩子，仅在企业版中支持。
Python 和 Java 驱动基于 Erlang/OTP-Port 进程间通信实现，本身具有非常高的吞吐性能，本文以 Python 拓展为例介绍 EMQ X 跨语言拓展使用方式。
Python 拓展使用示例
要求
EMQ X 所在服务器需安装 Python 3.6 以上版本
使用步骤
通过 pip 安装 Python SDK
调整 EMQ X 配置，确保相关配置项正确指向 Python 项目
引入 SDK 编写代码
Python 插件安装
通过 pip 命令在本地安装 SDK，确保使用 pip3 进行安装：
pip3 install emqx-extension-sdk
修改配置
修改 emqx-extension-hook 插件配置，正确使用拓展：
## Setup the supported drivers#### Value: python2 | python3 | javaexhook.drivers = python3## Search path for scripts/libraryexhook.drivers.python3.path = data/extension/hooks.py## Call timeout#### Value: Duration##exhook.drivers.python3.call_timeout = 5s## Initial module name## Your filename or module nameexhook.drivers.python3.init_module = hooks
编写代码
在 emqx/data/extension 目录下新建 hooks.py 文件，引入 SDK 编写业务逻辑，示例程序如下：
## data/extension/hooks.pyfrom emqx_extension.hooks import EmqxHookSdk, hooks_handlerfrom emqx_extension.types import EMQX_CLIENTINFO_PARSE_T, EMQX_MESSAGE_PARSE_T# 继承 SDK HookSdk 类class CustomHook(EmqxHookSdk): # 使用装饰器注册 hooks @hooks_handler() def on_client_connect(self, conninfo: EMQX_CLIENTINFO_PARSE_T = None, props: dict = None, state: list = None): print(f'[Python SDK] [on_client_connect] {conninfo.clientid} connecte') @hooks_handler() def on_client_connected(self, clientinfo: EMQX_CLIENTINFO_PARSE_T, state: list = None): print( f'[Python SDK] [on_client_connected] {clientinfo.clientid} connected') @hooks_handler() def on_client_check_acl(self, clientinfo: EMQX_CLIENTINFO_PARSE_T, pubsub: str, topic: str, result: bool, state: tuple) -> bool: print( f'[Python SDK] [on_client_check_acl] {clientinfo.username} check ACL: {pubsub} {topic}') # 用户名为空时，ACL 验证不通过 if clientinfo.username == '': return False return True @hooks_handler() def on_client_authenticate(self, clientinfo: EMQX_CLIENTINFO_PARSE_T, authresult, state) -> bool: print( f'[Python SDK] [on_client_authenticate] {clientinfo.clientid} authenticate') # clientid 不为空时，验证通过 if clientinfo.clientid != '': return True return False # on_message_* 仅支持企业版 @hooks_handler() def on_message_publish(self, message: EMQX_MESSAGE_PARSE_T, state): print( f'[Python SDK] [on_message_publish] {message.topic} {message.payload}')emqx_hook = CustomHook(hook_module=f'{__name__}.emqx_hook')def init(): return emqx_hook.start()def deinit(): return启动
启动 emqx_extension_hook 插件，如果配置错误或代码编写错误将无法正常启动。启动后尝试建立 MQTT 连接并观察业务运行情况。
./bin/emqx_ctl plugins load emqx_extension_hook
感谢各位的阅读，以上就是"怎么用Python开发EMQ X MQTT服务器插件"的内容了，经过本文的学习后，相信大家对怎么用Python开发EMQ X MQTT服务器插件这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是，小编将为大家推送更多相关知识点的文章，欢迎关注！
