公司最近与第三方合作的一款硬件产品需要每隔三秒请求服务器上报设备状态， 并且每隔十秒可能需要通过服务器获取比较大量的数据到设备端进行运算。 这对服务器的并发能力提出了很大的挑战。 目前服务端使用的框架是Django 1.11.4版本，无法支持ASGI。 因此需要寻找一款并发性能可以满足需求且易于使用的ASGI框架。 由于团队成员之前一直使用的框架都是Django，大家对Django都很熟悉。 Django ORM与Django Rest Framework提供的很多特性也可以让开发同学快速高效的写出 安全的代码。所以一开始倾向与将Django升级到3.X版本，这样可以使用Django Channel。 不过，在尝试使用Django Channel框架实现几个异步接口之后，发现Django Channel在使用ORM进行 数据库操作的时候，需要使用channels.db.database_sync_to_async装饰器对数据操作方法进行封装。 这一点让人感觉有点不够优雅，并且实际测试下来发现性能也并没有很大的提升，于是放弃使用Django 3.X版本的想法。 之后，又使用Falcon ASGI框架实现接口进行测试，实际测试发现Falcon ASGI的并发性能相对与Django Channel 有很大的提升，并且代码实现也相对要简洁很多。这一点深得团队成员的喜爱，也很符合Python哲学。 Simple is better than complex 测试方法说明： 实现两个接口，分别返回不同长度数据，接口A返回100B，接口B返回200KB。 在本地通过脚本实现多客户端同时并发请求上述两个接口，脚本动态配置参数需求（以下参数需支持动态配置）： 模拟时长，精度：分钟；默认配置：10分钟。 模拟的同时生效的客户端个数；默认配置：500。 接口轮询的间隔，精度到毫秒；默认配置： 500ms 两个接口的请求占比；默认配置：10:1 结果输出需求： 以客户端为单位，分别统计两个接口的请求结果（成功/失败，最好能看到错误原因，但不强求）、请求耗时（可包含网络时间） 脚本执行中，服务器的资源占用率 测试结果: 模拟客户端请求代码: # -*- coding: utf-8 -*- import logging import os import threading import time from concurrent.futures import ThreadPoolExecutor import click import redis import requests logging.basicConfig( level=logging.INFO, format='%(levelname)s %(asctime)s %(threadName)s [%(lineno)s] %(message)s', datefmt="%Y-%m-%d %H:%M:%S", filename='fw_client.

最近有个功能需要使用gpg对服务器为不同工厂生成的OTP文件加密，且每个工厂的gpg密钥都不固定。 由于服务是运行在容器内的，所以需要在执行加密的前导入配置的gpg密钥，并将其改成可以可信任的。 gpg在导入密钥后，修改为可信任度时需要在指令执行过程中与shell有几次交互过程，所以简单的 os.system(cmd)无法完成操作。 通过一番资料查阅，最终找到了pexpect，可以完美的解决在执行shell命令的过程中需要交互的问题。 示例代码如下： import pexpect # 导入工厂的gpg密钥 gpg_key_file = "F2A3D51D6164AA724461D02D26BF344D2FEC2866.key" child = pexpect.spawn("gpg --import {}".format(gpg_key_file)) print(child.read()) child.close() # 修改gpg密钥的信任等级 gpg_key_name = "F2A3D51D6164AA724461D02D26BF344D2FEC2866" child = pexpect.spawn('gpg --edit-key {}'.format(gpg_key_name)) child.expect("gpg") child.sendline('trust') child.sendline('5') child.sendline('y') child.sendline("quit") child.close() 参考资料 GPG入门教程 Pexpect Document

Python Regex Example Record 从日志文本中提取字典结构数据 import re import json s = '''[2023-04-19 10:18:36,070] [73709] [INFO] {"request": "GET /hello/?format=json HTTP/1.1", "status": "200", "bytes": "30", "remote_address": "127.0.0.1","duration": "0.883924", "referer": "-", "agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/112.0.0.0 Safari/537.36"}''' pattern = r"{.+[:,].+}" items = re.findall(pattern, s) for item in items: print(json.loads(item))