最近要支持一个功能，让用户自定义一些动态的逻辑，索性让用户自己写代码执行，所以需要做到让用户的代码动态执行起来。下面列举一下核心逻辑。 代码 1 2 3 4 5 6 7 8 9 10 11 12 13 14 #!/usr/bin/env python3 import importlib def new_module(mod_name): spec = importlib.machinery.ModuleSpec(mod_name,None) return importlib.util.module_from_spec(spec) def create_module(mod_name, object_list): mod = new_module(mod_name) for obj in object_list: setattr(mod,obj.__name__, obj) return mod 这个脚本主要是用来动态创建module，并且把对象放到module中，这样就可以在其他地方调用了。 参考文章 Creating Dynamic Modules in Python 3 ……

阅读全文

背景 近期看到企业微信告警，发现线上服务进程隔一段时间就会重启一次，遂开始进行排查。 线上服务描述 Python服务(使用aiohttp) K8S部署 容器中使用gunicorn，每个容器启动4个worker进程 排查过程 观察报警信息，只有一部分线上环境收到了影响。 观察报警时间，观察到只在整点时有一次服务重启告警(告警是进程自己发送的，也就是说只有一个进程重启了)，并且不是必现。 观察资源信息，在服务重启前，重启信息的容器的内存有一次跳跃式增长。 最终，总结出几个信息: 某一个环境(该环境数据量大)，整点时分，单进程重启，内存暴涨造成的。 而服务在整点时只有一个定时任务，这个定时任务是做多数据库的数据同步。 并且，任务的调起，是由外部的一个服务向本服务发送的一个HTTP请求（只会有一个进程来处理这个HTTP请求），所以，能够说通为什么只有一个进程会重启。 观察该HTTP请求的信息，查看该请求日志中的进程ID和重启的进程ID，最后发现是一致的，并且重启时间的定时任务日志中，没有定时任务完成得日志数据，所以最终推断出进程得重启是由该定时任务导致内存暴涨，从而导致进程重启。 但，还有一个问题没有被印证，进程重启的问题不是必现的，进而，我们需要拉长时间观察，最终发现，进程会因为定时任务导致内存有一次跳跃式的增长，但不一定会导致OOM，但有一个问题，内存虽然增加了，但是并没有看到内存回收，这个需要了解一下Python的内存管理机制，是因为进程执行过程中没有一个Arena中的内存被释放完，所以就不会被回收。 解决问题 通过多进程的方式 从上文的排查过程来看，我们可以从内存回收的角度来，在每次定时任务后保证申请的内存能够还给操作系统，避免造成Python一直持有内存不归还。但直接优化GC和内存管理对我一个普通的小开发来说不太现实，所以换个角度看，进程死亡，他的内存一定会归还，那这样就可以绕开Python内存管理造成的内存不归还的问题了。所以，我尝试了以下办法： 直接在http服务进程中做fork操作，但在aiohttp中执行fork视乎会出现各种各样的小问题，例如，fork出去的进程会copy一份父进程的内存，相当于和父进程使用了同一份async loop对象，在子进程中，会报错loop是异常的报错。所以我猜测，Python异步和同步fork操作应该不同。 在GitHub中查到了一个叫aiomultiprocess的三方库解决了办法1的问题，不会报出loop异常的问题。但，我开发时是未使用gunicorn的，所以在测试环境测试时，发现fork操作其实会卡死，很是困惑，遂Google到一篇文章Gunicorn+GeventWorker环境下fork进程意外结束的问题，这篇文章的解决办法有些trick，不太想用，放弃。 既然fork两条路都走不通，再想出了一个新的方式，脚本执行。我将定时任务中的逻辑，不再放到HTTP请求里面了，而是重新修改为脚本，在HTTP调用中，调起这个脚本。最终发现，行得通，上线。 上线后，发现内存确实是能够正常回收了，内存跳跃式增长不回收，变成了一个又一个的内存尖刺。但过了几周后，又发现了服务重启，查看原因是脚本占用内存太大，导致某个服务进程在申请内存时OOM，虽然告警的频率极低，但也算是一个隐患，需要优化它。 优化代码逻辑，降低内存消耗 从上文中看到，虽然解决了内存回收的问题，但随着业务的日益增长，数据量也会增长，数据同步逻辑会随着数据量的增长导致内存增长，而影响了服务进程。但是由于某些原因，不能新开一个服务来做这个事情，所以，查看了数据同步逻辑的代码，大概逻辑是：对比一个时间段内源数据库与目标数据的数据数量，如果相等，不迁移数据，如果不相等，同步数据。但同步数据并不是按照差异同步，而是使用了源数据库单时间段内的数据全量覆盖。有些暴力了，并且单看数据数量就判断是否需要同步数据也有失偏颇。 所以，最终修改为，根据最后更新时间来对比数据，同步有差异数据。 总结 解决本次的问题，从技术上来说，学习到了： Python的内存管理 了解了Gunicorn fork进程的一些机制 Python异步直接fork子进程的一些坑 从经验上看： 排查该类问题的一些思路以及逻辑。 对解决问题的思考，或者说对无法解决的问题，我如何的避开它，也就是如果规避一个问题的发生。 对数据同步，或者数据对比一些业务逻辑的思考。 ……

阅读全文

GIL 保证的 Thread-safe 是在 Bytecode 层而不是 Python Code。所以能确保的是每行 Bytecode 都会被运行完成，而多行 Bytecode 是则有被中断切换执行的可能性。 原子操作 读取/覆盖一个单变量 读取/覆盖一个全局变量 从list拿到一个元素 更新list的一个元素(例如: list.append()) 从dictionary中获取一个元素 更新dictionary的一个元素 参考文章 Atomic operations ……

阅读全文

GIL是什么 首先需要明确的一点是GIL并不是Python的特性，它是在实现Python解析器(CPython)时所引入的一个概念。就好比C++是一套语言（语法）标准，但是可以用不同的编译器来编译成可执行代码。有名的编译器例如GCC，INTEL C++，Visual C++等。Python也一样，同样一段代码可以通过CPython，PyPy，Psyco等不同的Python执行环境来执行。像其中的JPython就没有GIL。然而因为CPython是大部分环境下默认的Python执行环境。所以在很多人的概念里CPython就是Python，也就想当然的把GIL归结为Python语言的缺陷。所以这里要先明确一点：GIL并不是Python的特性，Python完全可以不依赖于GIL。 那么CPython实现中的GIL又是什么呢？GIL全称Global Interpreter Lock为了避免误导，我们还是来看一下官方给出的解释： In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple native threads from executing Python bytecodes at once. This lock is necessary mainly because CPython’s memory management is not thread-safe. (However, since the GIL exists, other features have grown to depend on the guarantees that it enforces.) 好吧，是不是看上去很糟糕？一个防止多线程并发执行机器码的一个Mutex，乍一看就是个BUG般存在的全局锁嘛！别急，我们下面慢慢的分析。 为什么会有GIL 由于物理上的限制，各CPU厂商在核心频率上的比赛已经被多核所取代。为了更有效的利用多核处理器的性能，就出现了多线程的编程方式，而随之带来的就是线程间数据一致性和状态同步的困难。即使在CPU内部的Cache也不例外，为了有效解决多份缓存之间的数据同步时各厂商花费了不少心思，也不可避免的带来了一定的性能损失。 Python当然也逃不开，为了利用多核，Python开始支持多线程。而解决多线程之间数据完整性和状态同步的最简单方法自然就是加锁。 于是有了GIL这把超级大锁，而当越来越多的代码库开发者接受了这种设定后，他们开始大量依赖这种特性（即默认python内部对象是thread-safe的，无需在实现时考虑额外的内存锁和同步操作）。 慢慢的这种实现方式被发现是蛋疼且低效的。但当大家试图去拆分和去除GIL的时候，发现大量库代码开发者已经重度依赖GIL而非常难以去除了。有多难？做个类比，像MySQL这样的“小项目”为了把Buffer Pool Mutex这把大锁拆分成各个小锁也花了从5.5到5.6再到5.7多个大版为期近5年的时间，并且仍在继续。MySQL这个背后有公司支持且有固定开发团队的产品走的如此艰难，那又更何况Python这样核心开发和代码贡献者高度社区化的团队呢？ 所以简单的说GIL的存在更多的是历史原因。如果推到重来，多线程的问题依然还是要面对，但是至少会比目前GIL这种方式会更优雅。 GIL 到底锁的是什么？ 大家都说 Python 有 GIL 锁，那么这个锁到底锁的是什么东西？？……

阅读全文

步骤 打开pydantic/main.py文件并且搜索ModelMetaclass定义，在ModelMetaclass类型定义，粘贴以下代码: 1 2 3 4 5 6 7 8 def __dataclass_transform__( *, eq_default: bool = True, order_default: bool = False, kw_only_default: bool = False, field_descriptors: Tuple[Union[type, Callable[..., Any]], ...] = (()), ) -> Callable[[_T], _T]: return lambda a: a 增加包装器在ModelMetaclass定义上: 1 @__dataclass_transform__(kw_only_default=True, field_descriptors=(Field, FieldInfo)) 结果 修改前 修改后 参考文献 ……

阅读全文

问题 在Python3中，例如：构建链表节点类，会有一个指向自身类型的指针，代码如下: 1 2 3 4 5 class ListNode(object): value: int nextNode: ListNode def merge(self, head: ListNode): pass 如果直接写成这样，代码提示器是无法识别到的 解决 Python 3.10以及以后的版本 已经支持该方式 Python 3.7+ 使用feature 1 from __future__ import annotations Python 3.6以及更旧的版本 直接使用字符串的方式，例如: 1 2 3 4 5 class ListNode(object): value: int nextNode: 'ListNode' def merge(self, head: 'ListNode'): pass 参考资料 https://stackoverflow.com/questions/33533148/how-do-i-type-hint-a-method-with-the-type-of-the-enclosing-class……

阅读全文

今天遇到线上一个BUG，用户导入数据时，将某个字段填写一个负数会导致服务器报错，大概的代码如下: 1 2 3 4 if isinstance(obj, str) and obj.isdigit(): obj = int(obj) else: raise Exception("fuck you") 但是，负数并不能直接使用isdigit判断，看下截图： 所以，还需要做一些转换，满足这个需求: 1 2 3 4 5 6 7 8 9 10 def isdigit(num: str) -> bool: if not num: return False if num[0] == '-': return num[1:].isdigit() return num.isdigit() if isinstance(num, str) and isdigit(num): num = int(num) else: raise Exception("fuck you") ……

阅读全文

三方组件 paramiko 创建ssh 连接 1 2 3 4 5 6 7 8 9 10 11 12 13 def connect_ssh(host: str, port: int, username: str, password: str): ssh = paramiko.SSHClient() key = paramiko.AutoAddPolicy() ssh.set_missing_host_key_policy(key) ssh.connect(host, port, username, password, timeout=5) return ssh 执行命令 1 2 3 4 5 6 7 8 9 def execute(ssh: paramiko.SSHClient): stdin, stdout, stderr = ssh.exec_command('command') for line in stdout.readlines(): print(line) ……

阅读全文

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 #!……

阅读全文

使用__new__方法 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 class Singleton(object): def __new__(cls, *args, **kw): if not hasattr(cls, '_instance'): orig = super(Singleton, cls) cls._instance = orig.__new__(cls, *args, **kw) return cls._instance class MyClass(Singleton): pass 共享属性 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 class Borg(object): _state = {} def __new__(cls, *args, **kw): ob = super(Borg, cls).……

阅读全文