【昉·星光2】Python MQTT测试

分享作者：AuroraTea

评测品牌：赛昉科技

评测型号：VF202040-A0

发布时间：2025-07-18 10:01:28

0 0

视频链接

https://blog.csdn.net/qq1352685369/article/details/149437856

前言

本文介绍了在昉星光2（RISC-V架构）上运行Linux发行版和Python开发的体验。作者测试了新一代Python包管理工具uv，其性能优于传统工具且支持依赖共享。通过分析官方安装脚本，确认了其对RISC-V的支持。重点展示了基于MQTT v5协议的请求-响应模型实现，使用HiveMQ Broker进行跨架构测试（RISC-V vs ARM），验证了RISC-V在物联网应用中的可行性。

开源口碑分享内容

昉星光2支持运行完整的Linux发行版操作系统, 所以可以直接运行Python等高级语言. HTTP, MQTT等操作非常轻松便捷, 也可以直接使用纯软件开发的包管理工具, 比如说pip/uv, 开发体验非常好.

我甚至用上了比较新的Python包管理工具uv, 它其实不仅仅是包管理--我原本使用Portey做包管理和虚拟环境 , pyenv做Python本身的版本管理, 而现在一个uv就可以完成所有的这些工作, 而且速度提升确实非常明显, 同时它下载的依赖类似node的pnpm, 不同项目间同样的依赖只是多一份软链接, 不会反复占用空间.

虽然在官网中我无法找到RISC-V或者risc或者riscv相关的内容,

但是在官方的安装脚本https://astral.sh/uv/install.sh中, 还是可以找到对riscv的支持, 所以也是放心运行curl -LsSf https://astral.sh/uv/install.sh | sh进行安装.

        "riscv64gc-unknown-linux-gnu")
            _archive="uv-riscv64gc-unknown-linux-gnu.tar.gz"
            if ! check_glibc "2" "31"; then
                _archive=""
            fi
            if [ -n "$_archive" ]; then
                echo "$_archive"
                return 0
            fi
            ;;

我准备测试一下物联网中非常经典的MQTT操作., 并且实现一个用MQTT这种发布-订阅的形式去实现一个请求-响应模型, 同时MQTT v5对此本身也有提供更便捷的支持. 注意其中提到的Server和Client是相对于请求-响应模型而言的, 对于底层的MQTT来说他们都是Client(例如Server的代码中import paho.mqtt.client as mqtt就很明显, 我变量名和参数名的命名规则也是保持一致), Broker是client.connect时连接的对象.

Client:

import paho.mqtt.client as mqtt
from paho.mqtt.subscribeoptions import SubscribeOptions
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes
import time
 
def on_message(client, userdata, message):
    print ('response received at client')
    print (message.payload)
 
client = mqtt.Client(client_id='client', protocol=mqtt.MQTTv5)
client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
client.username_pw_set("test_client_username", "password")
client.connect("suburl.s1.eu.hivemq.cloud", 8883, 10)
 
client.subscribe('common', options=SubscribeOptions(noLocal=True))
 
client.on_message = on_message
 
client.subscribe('response')

publish_properties = Properties(PacketTypes.PUBLISH)
publish_properties.ResponseTopic = 'response'
publish_properties.CorrelationData = b"334"
 
client.publish('common', "client message", 1, properties=publish_properties)
print ('request sent from client')
 
client.loop_start()
time.sleep(20)
client.loop_stop()

Server:

import paho.mqtt.client as mqtt
from paho.mqtt.subscribeoptions import SubscribeOptions
import time
 
def on_message(client, userdata, message):
    try:
        print ("request recieved at server")
        print (message.payload)
        resopnse_topic = message.properties.ResponseTopic
        client.publish(resopnse_topic, 'server message',1, properties=message.properties)
        print ('response sent from server')
    except Exception:
        print('error')

client = mqtt.Client(client_id='server', protocol=mqtt.MQTTv5)
client.tls_set(tls_version=mqtt.ssl.PROTOCOL_TLS)
client.username_pw_set("test_server_username", "password")
client.connect("suburl.s1.eu.hivemq.cloud", 8883, 10)
 
client.subscribe('common', options=SubscribeOptions(noLocal=True))
 
client.on_message = on_message
client.loop_start()
time.sleep(100)
client.loop_stop()

目前网上可以免费进行测试的MQTT Broker提供商非常多, 这次我是使用HiveMQ来进行测试的, 国内也比较推荐EMQX, 基本上都只是在tls_set / username_pw_set / connect处填写的参数有所不同, 都用MQTT标准客户端即可, 我这用的就是paho (Python的paho-mqtt). 整体所用代码逻辑基本和我以前在ARM的Linux上写的一致, 当时经过测试是完全ok的, 而现在在RISC-V的Linux上运行的结果也正常, 说明在这种场景RISC-V已经可以替代ARM了, 期待推出更多的产品.

全部评论

暂无评论

0/144