说明

# 主题
python 语言特性

# 介绍
python 的个人使用经验；
总结了 python 核心特性，按照概念、总结、示例的顺序逐步说明；
不涉及 python 源码实现，均为 python 代码示例；

对象

• 概念

# 对象
python 中的一切都是对象：
- 所有数据都是对象
- 所有类型本身也是对象
- 函数、类、模块甚至代码块也是对象

python 中的任何东西都可以被赋值给变量、作为参数传递、或者作为返回值返回；

# 引用
所有的变量都是对象的引用（类似别名），不直接存储对象（不会涉及内存操作）；

• 总结

在做大型项目的开发时，需要考虑对象的生命周期以及读写控制；

* 生命周期
只要引用存在，对象就不会回收（循环引用除外）；
【建议】尽量避免使用全局变量；
  
* 读写控制
所有变量引用的都是同一个对象，容易修改；
【建议】尽量提供接口控制，合理规划读写场景；

• 示例

# 示例：内置方法

a = 1
assert type(a) is int # 查看实例的类
assert a is a  # 判断是否为同一个对象
print(dir(a))  # 查看实例上定义的属性和方法
print(id(a))  # 进程内唯一标识（虚拟地址）
# 示例：将函数作为参数传递，并作为返回值返回

def a():
    pass

def b(arg):
    return arg

assert b(a) is a  # 同一个对象

# 示例：将函数赋给变量
c = b
assert b is c # 同一个对象

assert c(a) is a # 等价于 b
# 示例：返回函数内的对象，在函数外使用

id1 = None

def c():
    global id1
    s = 'a_string'
    id1 = id(s)
    return s 

assert id(c()) == id1  # 同一个对象

多态

• 概念

# 多态
多态是一种编程语言能力，将算法与类型解耦，提高代码复用性；

通用分类：
- 函数重载
- 继承/接口
- 泛型（编译时类型参数化）与模板（编译时生成代码）

python 实现：
通过鸭子类型实现多态，不需要类型定义，只需要运行时行为；
具体而言，不在编译阶段检查对象类型，而是在运行时，判断对象是否具备执行的能力；

# 类型注释
背景：由于 python 的多态能力，编写 python 代码时对象是不需要定义类型的，增加了理解代码的难度；

通过类型注释可以标注类型，类型注释不影响代码执行逻辑；

• 总结

python 的多态不需要提前定义接口或抽象类，实现起来语法简单；

类型注释可以触发静态语法检查，加强 ide 功能，包括：
- 校验参数类型，例如，对象属性及方法的访问和使用
- 编码过程自动触发语法提示

• 示例

# 示例：多态实现

# 1. 定义出处理逻辑
def func(obj):
    res = obj + 1
    res.double()
    return str(res)

# 2. 实现满足该处理逻辑/行为的对象
class A:

    def __init__(self, data):
        self.data = data

    def __add__(self, other):
        # 重载运算符
        self.data += 1
        return self  # 返回同一个对象

    def double(self):
        self.data *= 2

    def __str__(self):
        # 重载 str 函数
        return f'data: {self.data}'

a = A(1)
assert func(a) == 'data: 4'
# 示例：类型注释

import typing

class A:
    
    def __init__(self, data):
        self.data = data

# 无类型注释
# 编写过程中没有任何提示，编写完成后能理解部分含义；
def func1(mapping):
    for k, v in mapping.items():
        print(k, v.data)

# 有类型注释
# 编写过程提供语法提示；编写完成后提供静态语法检查；代码阅读时可以跳转；
def func2(mapping: typing.Dict[str, A]) -> None:
    for k, v in mapping.items():
        print(k, v.data)

m = {
    'x': A(1),
    'y': A(2),
}
# 静态语法检查会检查传参类型
func2(m)

内存管理

• 概念

# 内存分配
python 中一切皆为对象，创建对象以及回收对象都是自动管理的，不需要显式处理；

# 垃圾回收
引用计数：对象有新引用时加一，引用它的对象被删除时减一，引用计算为 0 时回收；
循环引用：通过遍历所有对象，不可达对象为循环引用，回收；
分代回收：根据对象存活时间，触发回收逻辑（存活越久，回收频率越低）；

# 生命周期
一般回收对象主要考虑引用计数为 0 的场景，包括：
- 函数结束后，销毁局部变量
- 变量不再指向原对象
- 删除引用（del）
- 程序结束

• 总结

简单来说，对象的生命周期，就是从创建该对象开始，到删除对它的所有引用；

• 示例

# 示例：生命周期

def func1():
    # 该字典对象持续到函数结束 
    a = dict()
    a['x'] = 1
    print(a)

def func2():
    a = dict()
    a['x'] = 1
    
    # 替换变量引用的对象，导致上一个对象的引用计数清零，被回收
    a = set()
    print(a)

模块

• 概念

# 代码组织方式
module：单个 py 文件
package：包含一个或多个 py 文件的文件夹（可嵌套）

# 启动 python 程序
执行 python main.py 时：
1. 操作系统启动 python 解释器进程
2. 创建一个主线程和一个解释器实例
3. 解释器读取并编译 main.py
4. 从上到下执行该模块 main 的所有顶层代码

顶层代码包括：
import、类定义、函数定义、全局变量、其它代码

# 模块
文件层面：一个 py 文件；
运行层面：执行 import 流程的结果，是一个 module 类型的实例；

# import 流程
import 会加载模块，执行模块顶层代码，生成 module 对象，并绑定到当前命名空间； 
- 导入文件会执行/编译文件内容；
- 导入文件夹会执行/编译 __init__.py 文件内容；

本地模块的通用导入流程：
> 根据导入名称查找内存，如果有缓存的模块对象，则直接返回（在当前命名空间创建变量引用该模块对象）
> 根据搜索路径（文件夹）查找模块（文件/文件夹）
> 如果没有模块则拋出导入异常，中断导入流程
> 如果模块有字节码缓存文件，直接加载并返回
> 执行加载过程，将模块对应的 py 文件编译为字节码，生成字节码缓存文件，创建模块对象写入内存，返回

• 总结

# 模块及模块全局变量
模块导入后会缓存在内存中，重复 import 会尝试从缓存中读取已加载的模块，所以只会执行一次导入流程；
模块中创建的对象（全局变量），生命周期是从导入模块开始，到程序结束的；

# 线程安全
模块是天然的单例模式（顶层代码只会执行一次且缓存在内存中），在多线程中使用模块级别的变量，需要考虑并发问题（线程安全）；
跨线程使用的模块，需要确认模块是否线程安全；

• 示例

# 示例，模块级别定义的变量

# -----------------
# code1.py 文件内容如下：
# -----------------

# 导入模块会执行从上到下执行所有顶层代码

# 顶层代码：全局变量
x = {}

# 顶层代码：类定义
class A(object):

    # 类定义包括类变量，所以类变量也是全局变量
    data = {}
    
    def log(self):
        print(self.__class__.data)

# 顶层代码：函数定义
def func():
    pass

# 顶层代码：其它代码；
# 由于模块的缓存机制，重复导入该模块也只会执行一次
print('import one')
# ---------------------
# code2.py 文件内容如下：
# ---------------------
import code1

# 模块可以重复导入，但是只会执行一次；
import code1

print(code1.x)
print(code1.A.data)
# ---------------------
# code3.py 文件内容如下：
# ---------------------

# 同一个进程中导入相同的模块，访问的是相同的模块对象；
import code1
import code2

# code2 访问的对象和 code3 访问的是同一个 code1 模块
print(id(code1))
print(id(code2.code1))
# 示例，线程安全的模块，logging

import logging

# logging 模块中全局的顶层 logger
print(logging.root)

# logging 模块中全局的 manager
print(logging.root.manager)

# logging 对共享资源加锁
# logging/__init__.py:l217

# logging 模块中提供的配置函数，可以配置全局默认的日志处理器
# 加锁，配置 root logger 的 handler，最后释放锁
logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] [%(levelname)s] [%(name)s] [%(module)s.%(funcName)s] %(message)s',
)

# [2026-03-07 15:10:01,800] [INFO] [root] [test.<module>] info 2
logging.info('info')

进程与线程

• 概念

# 进程与线程
通用概念：
- 进程是资源分配的基本单位；
- 线程是调度执行的基本单位；

# 性能
gil（global interpreter lock），全局解释器锁是进程持有的，线程执行代码前需要先获取 gil；
一个进程可以有多个线程，由于 gil，同一时刻只允许一个线程执行 python 字节码，只能有效利用一个 cpu 核心；

io 操作会释放 gil 锁，所以线程能够处理 io 密集型任务，不适合 cpu 密集任务； 
现在 io 密集型任务一般使用异步 io 来实现；

# 资源
进程中的资源是线程共享的，如果是多线程服务，则需要考虑并发问题（对共享资源加锁）；
多进程的资源是互相独立的，如果创建多进程服务，需要考虑资源共享和通信问题；

# 线程安全
线程安全：
多线程并发访问同一份数据时，程序能够正确运行，不会产生错误结果；

基本概念：
- 临界区：多个线程并发访问，并且至少有一个线程会修改的代码区域
- 竞态条件：依赖于执行顺序，会产生不同的结果

实现线程安全的办法，核心原理是破坏并发读写条件

* 限制并发：对象不跨线程访问
- 加锁，限制临界区的访问
- thread local
- （不访问/引用全局变量的）局部变量，例如，工具包
- 通过线程安全的对象传递，限制同一时刻只有一个线程在访问，例如，生产消费模型中的队列

* 限制读写：对象只读
- 代码中约定只有读操作，例如，在多线程服务启动前准备好对象，不再修改
- 对象本身不可修改，限制了只读，例如，元组、字符串等等

• 总结

python 中的多进程在不同平台（windows/linux）上实现不同，非必要不使用多进程并发；
即使是多进程服务，一般也是框架实现，自己编写多进程服务的应用场景比较少；

python 进程默认只有一个主线程，如果需要启动新的线程，就需要考虑并发问题（线程安全等等）；
python 模块不一定都是线程安全的，也不一定都要求线程安全；

现有数据中心接触的业务，基本都是 io 密集型，优先选用多线程实现并发：

• 示例

# 示例，多线程应用中的配置管理：约定对象只读，不需要加锁

import threading

def load_configure():
    # 模拟配置加载过程
    return {0: {}, 1: {}, 2: {}}

def handler(cf, i):
    # 只读
    print(cf.get(i))

def main():
    # 提前加载配置
    configure = load_configure()

    ts = []
    for i in range(3):
        ts.append(threading.Thread(target=handler, args=(configure, i)))

    for t in ts:
        t.start()

    for t in ts:
        t.join()
# 示例，定时任务框架 apscheduler 配置，使用线程池执行器

# apscheduler 同时支持线程池与进程池执行器，当前使用线程池执行器；

gconfig = {
    'apscheduler.jobstores.default': {
        'type': 'memory',  # 无持久化
    },
    'apscheduler.executors.default': {
        'class': 'apscheduler.executors.pool:ThreadPoolExecutor',  # 线程池
        'max_workers': '20',  # 定时任务最大并发
    },
    'apscheduler.job_defaults.max_instances': '1',  # 同一个任务最大并发执行数量
    'apscheduler.timezone': 'Asia/Shanghai',  # 时区
}

生成器

• 概念

# 生成器
生成器是 python 中的一种惰性迭代机制，用于按需生成数据，而不是一次性生成全部数据。
生成器也是迭代器（实现了迭代器协议）；
生成器基于 yield 语法；

• 总结

基于生成器可以实现一些内存友好的代码，例如流式处理大数据；

• 示例

# 示例，生成器的逐步执行机制

# 方式一：通过 next 方法单步执行

def gen():
    yield 1
    yield 2
    yield 3

g = gen()
assert next(g) == 1  # 执行到 yield value 后暂停
assert next(g) == 2
assert next(g) == 3

# 生成器执行完成后，拋出异常
try:
    next(g)
except StopIteration:
    pass
# 示例，生成器的逐步执行机制

# 方式二：通过调用生成器的 send 方法单步执行

def gen():
    val = yield 1
    print(val)  # first
    val = yield 2
    print(val)  # second
    val = yield 3
    print(val)  # third

# next(g) 方法相当于 g.send(None)
g = gen()
assert g.send(None) == 1  # 启动生成器只能传参 None，并且忽略 None 值
assert g.send('first') == 2
assert g.send('second') == 3
try:
    g.send('third')
except StopIteration:
    pass
# 示例，生成器也是迭代器

def iterator(n):
    # [0, 1, ..., n-1]
    return list(range(n))

def generator(n):
    for i in range(n):
        yield i

def run():
    n = 3
    
    # 生成器也实现了迭代器协议
    for i in generator(n):
        print(i)
    
    for i in iterator(n):  
        print(i)
# 示例：从大文件中读取数据，流式处理

def load1(file):
    """读取文件：逐行读取文件，通过列表返回"""
    # open 读取文件提供了逐行读取的方式，避免一次性加载整个文件到内存中
    with open(file, 'r', encoding='utf-8') as f:
        data = []
        for line in f:
            data.append(line)
    return data

def load2(file):
    """读取文件：逐行读取文件，并通过生成器返回"""
    with open(file, 'r', encoding='utf-8') as f:
        for line in f:
            yield line

def stats(file):
    """统计文件中单词的出现频率"""
    
    # 生成器也实现了迭代器协议，可以用 for 遍历
    # iterator = load1(file)
    iterator = load2(file)
    
    res = {}
    for line in iterator:
        words = line.strip().split(' ')
        for word in words:
            res[word] = res.get(word, 0) + 1
    return res

# 使用列表，该程序最大占用内存为，文件大小加上统计字典大小
# 使用生成器，忽略单行字符串的大小，该程序最大占用内存为，统计字典大小
# 示例，从数据库中流式读取数据，并流式处理

# pymysql：mysql 数据库驱动
import pymysql.cursors

def iter_select():
    # 流式获取数据，并且逐行返回，内存最大占用为单次批量数据大小
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='123456',
        database='test',
        cursorclass=pymysql.cursors.SSCursor,  # 指定流式游标
    )

    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM big_table")

            while True:
                # 流式游标：每次实时从数据库获取批量数据
                # 默认游标：一次性从数据库获取所有数据，在内存中分批后再逐批返回
                rows = cursor.fetchmany(1000)  # 每次拉 1000 条
                if not rows:
                    break

                for row in rows:
                    yield row  # 每次返回 1 条
    finally:
        conn.close()

def run():
    # 对于主程序而言，不用关系底层是如何获取数据的，支持遍历的方式
    for row in iter_select():
        print(row)  # 模拟处理函数 

异步

• 概念

# 协程
一般概念：
比线程更轻量的并发单位，在同一个线程内执行，可以中途暂停和恢复执行；
协程是协作式调度（区别于抢占式），自己决定什么时候挂起，让出执行权；

在 python 中：
协程基于生成器实现，提供暂停与恢复的功能；
协程本身不包含调度逻辑；

# 异步 io
* 异步 io
python 中的异步一般是指的异步 io；
单个线程内，基于协程和事件循环的并发模型，大部分代码是用户态的代码（非系统调用）；

* 协程对象
所有的函数被封装成一个协程对象，可以单步执行，提供挂起和恢复机制；

* 事件循环
io 事件循环中注册了所有的 io 读写事件，阻塞至事件发生，然后取出所有就绪的事件，逐个处理；

# 异步 io 处理逻辑
忽略定时调度、非 io 事件任务；

* 事件循环调度
step 1: 
- 逐个处理就绪队列中的任务
step 2: 
- 如果有 io 事件，阻塞至事件发生
- 如果没有 io 事件，结束事件循环
step 3:
- 某个事件发生，将所有就绪的事件对应的回调（可能有多个），放入就绪队列

* 处理流程
事件循环启动，主协程放入就绪队列，执行事件循环调度；

* 注意
所有的调度都在单个线程内，同一时刻只会有一个协程在执行；
io 操作由操作系统执行，不会影响当前线程执行；

# 异步 io 中的定时器
事件循环与定时器是分离实现的；
定时器中记录了未来时间点要触发的回调；

* 定时器与事件循环协作
事件循环会阻塞至下一个定时任务开始（计算 timeout）

• 总结

异步 io 执行过程是单线程的，不适合处理 cpu 密集的任务；

协程中的 io 操作必须使用支持事件循环的异步接口（awaitable），否则会阻塞事件循环，失去异步并发能力；
例如，异步 http 请求包 aiohttp、异步 mysql 驱动包 aiomysql 等等；

在异步代码中要注意避免 cpu 密集的任务，如果必须，可以尝试使用额外的线程池来处理；
# 示例，使用异步接口
import time
import asyncio

async def func1():
    """非异步 sleep"""
    time.sleep(1)  # 系统实现，不占用 cpu，但会阻塞线程

async def func2():
    """异步 sleep"""
    await asyncio.sleep(1)  # 用户代码实现，会挂起协程，不阻塞线程（被调度）

async def main():
    # await func1()
    await func2()

# 底层会启动一个事件循环，并执行主协程
asyncio.run(main())
# 示例，同步与异步 http 请求

# 同步请求：requests
import requests

def fetch(url):
    resp = requests.get(url)
    return resp.text

def main():
    html = fetch('https://httpbin.org./get')
    print(html)

main()
# 示例，同步与异步 http 请求

# 异步请求：aiohttp
import aiohttp
import asyncio

async def fetch(url):
    # 异步 with 上下文
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    html = await fetch('https://httpbin.org/get')
    print(html)

asyncio.run(main())