11603 字
30 分钟
python学习笔记
2026-05-07

1.python 基本命令#

python环境在conda中,使用anaconda prompt

1.使用conda创建python环境,安装python版本
conda create -n 环境名 python=3.8 # 创建python3.8环境
-n: 指定环境名称
python=3.8: 指定python版本
2.查看当前已有的环境
conda env list # 列出所有环境
3.激活环境
conda activate py38 # 进入python3.8环境
4.安装依赖包
conda install 包名 # 安装包
conda install -r requirements.txt # 从文件安装依赖包
5.查看当前已有的环境
conda env list # 列出所有环境
6.退出环境
conda deactivate # 退出当前环境
进入环境后,输入以下命令:

常用的python命令:

python -version # 查看python版本
python -m pip list # 查看已安装的包
python -m pip install 包名 # 安装包
python -m pip install -r requirements.txt # 从文件安装依赖包
python -m pip freeze > requirements.txt # 将当前环境的包列表输出到文件
python -m pip uninstall 包名 # 卸载包
python -m pip show 包名 # 显示包信息
python # 进入交互式命令行
exit() # 退出交互式命令行
python 文件名.py # 运行python文件

2. 基础语法#

2.1 注释与缩进#

python使用#进行单行注释,使用''' '''或""" """进行多行注释。 python使用缩进来表示代码块,通常使用4个空格(也可以使用tab,但不推荐)表示一个缩进级别。缩进必须一致,否则会导致IndentationError错误。 函数之间或类的方法之间用空行分隔,表示一段新的代码的开始。

分支语句中
if True:
print ("Answer")
print ("True")
else:
print ("Answer")
print ("False") # 缩进不一致,会导致运行错误
再比如循环中
for i in range(5):
print(i) # 这是循环体,必须缩进
print("循环内的代码") # 这是循环体,必须缩进
print("循环结束") # 这是循环外的代码,不需要缩进
多行语句
Python 通常是一行写完一条语句,但如果语句很长,我们可以使用反斜杠 \ 来实现多行语句,
total = item_one + \
item_two + \
item_three

2.2 变量与数据类型#

python是动态类型语言,变量不需要声明类型,直接赋值即可。 常见的数据类型有:整数(int)、浮点数(float)、字符串(str)、布尔值(bool)、列表(list)、元组(tuple)、字典(dict)等。

# 变量赋值
x = 10 # 整数
y = 3.14 # 浮点数
name = "Alice" # 字符串
is_valid = True # 布尔值
# 数据类型示例
my_list = [1, 2, 3, 4, 5] # 列表
my_tuple = (1, 2, 3) # 元组
my_dict = {"name": "Alice", "age": 30} # 字典
# 输出变量类型
print(type(my_tuple)) # <class 'tuple'>

为什么 Python 不需要声明变量类型?#

因为 Python 是动态类型语言,变量类型在运行时根据赋值自动确定,无需预先声明。

变量本身没有固定类型,类型属于它所引用的对象。

动态类型机制如何实现?#

Python 主要通过对象引用运行时类型信息实现动态类型:

  1. 变量是引用:执行 a = 42 时,Python 创建整数对象并让 a 指向它;执行 a = "Hello" 时,a 改为指向字符串对象。
  2. 类型属于对象:对象内部保存类型信息。CPython 中,PyObjectob_type 指针指向具体类型,如 PyLong_TypePyUnicode_Type
  3. 运行时动态派发:执行 a + b 时,解释器运行时检查对象类型,并查找相应的运算实现(如 nb_add__add__)。

注意:动态类型不等于没有类型,而是类型检查推迟到运行时。

详细介绍部分数据类型

2.2.1字符串#

与c++中的字符串类似,python中的字符串是不可变的,可以使用单引号、双引号或三引号来定义字符串。

str1 = "Hello, World!" # 双引号定义字符串
常用的字符串操作方法有:
str1.upper() # 转换为大写
str1.lower() # 转换为小写
str1.strip() # 去除首尾空格
str1.split(",") # 按照逗号分割字符串
str1.replace("World", "Python") # 替换字符串中的内容
str1.find("World") # 查找子字符串的位置,返回索引,如果找不到返回-1
str1.isalpha() # 判断字符串是否只包含字母,返回True或False
str1[0] # 获取字符串的第一个字符
#字符串切片
str1[start:end:step] #切片操作,获取字符串的子串,start表示起始索引,end表示结束索引(左闭右开),step表示步长,默认值为1
eg: str1[0:5] # 获取字符串的前5个字符,结果为 "Hello"
str1[7:] # 获取字符串从索引7开始到末尾的子串,结果为 "World!"
str1[-1] # 获取字符串的最后一个字符
str1[-5:] # 获取字符串的最后5个字符,结果为 "orld!"
str1[0:-1] # 获取从开始到倒数第二个字符的子串,结果为 "Hello, World"
str[1:5:2] # 获取字符串从索引1到索引5(不包含)的字符,步长为2,结果为 "el"
str1[::2] # 获取字符串的偶数索引字符,结果为 "Hlo ol!"
str1[::-1] # 获取字符串的反转,结果为 "!dlroW ,olleH"
#字符串输出
print(str1 * 2) # 输出字符串两次
print(str1 + '你好') # 连接字符串
print('hello\nrunoob') # 使用反斜杠(\)+n转义特殊字符
输出:hellorunoob
print(r'hello\nrunoob') # 在字符串前面添加一个 r,表示原始字符串,不会发生转义
输出:hello\nrunoob
# print默认换行输出,不换行输出需要在变量末尾加上 end="":
print( str1, end=" " )
print("这是下一条内容")
输出:Hello, World! 这是下一条内容

2.2.2 列表#

列表是python中最常用的数据类型之一,是一个有序的可变集合,可以包含不同类型的元素。列表使用方括号 [] 定义,元素之间用逗号分隔。

与c++与java中的数组不同,python中的列表可以动态调整大小,可以包含不同类型的元素,并且提供了丰富的方法来操作列表。类似于java中的ArrayList和c++中的vector,但更为灵活和强大。

# 创建列表
my_list = ['abcd', 786, 2.23, 'runoob', 70.2] # 避免使用 list 作为变量名,会覆盖内置类型
tinylist = [123, 'runoob']
print(my_list) # 打印整个列表:['abcd', 786, 2.23, 'runoob', 70.2]
print(my_list[0]) # 打印第一个元素(索引 0):abcd
print(my_list[1:3]) # 打印索引 1 和 2 的元素(不含索引 3):[786, 2.23]
print(my_list[2:]) # 打印从索引 2 开始到末尾的所有元素:[2.23, 'runoob', 70.2]
print(tinylist * 2) # 重复打印 tinylist 两次:[123, 'runoob', 123, 'runoob']
print(my_list + tinylist) # 拼接两个列表
#list常用操作
#1.访问列表元素
my_list[0] # 通过索引访问列表元素,索引从0开始,负数索引表示从末尾开始计数
#2.修改列表元素
my_list[0] = "new_value" # 通过索引修改列表元素
print(my_list) # 打印修改后的列表 [new_value, 786, 2.23, 'runoob', 70.2]
#3.添加元素
my_list.append("new_element") # 在列表末尾添加新元素
#4.删除元素
my_list.remove(786) # 删除列表中第一个匹配的元素
my_list.pop(0) # 删除指定索引的元素,并返回该元素,默认删除最后一个元素
#5.列表长度
my_list_length = len(my_list) # 获取列表的长度(元素个数) #len()是python内置函数,返回对象的长度或元素个数
#6.列表切片
#my_list[start:end:step] # 切片操作,获取列表的子列表,start表示起始索引,end表示结束索引(左闭右开),step表示步长,默认值为1
#7.列表排序
#my_list.sort() # 对列表进行升序排序
#warning 注意列表中的元素必须是可比较的类型,否则会抛出 TypeError 异常
#my_list.sort(reverse=True) # 对列表进行降序排序
#my_list_sorted = sorted(my_list) # 返回一个新的升序排序的列表,不改变原列表 #sorted()是python内置函数,返回一个新的排序后的列表
#8.列表反转
my_list.reverse() # 反转列表中的元素顺序
my_list_reversed =reversed(my_list) # 返回一个反转后的迭代器,不改变原列表 #reversed()是python内置函数,返回一个反转后的迭代器
print(str(list(my_list_reversed))) # 将迭代器转换为列表并打印
#9.元素计数
count = my_list.count('runoob') # 统计列表中某个元素的出现次数

2.2.3 元组#

元组(tuple)与列表类似,但元组是不可变的,一旦创建就不能修改。元组使用圆括号 () 定义,元素之间用逗号分隔。元组可以包含不同类型的元素。 如果与c++中的数组相比,元组更像是一个固定大小的数组,不能动态调整大小,但可以包含不同类型的元素。 那为什么有了列表还要有元组呢? 主要是因为元组的不可变性使得它在某些场景下更安全,比如作为字典的键,或者在多线程环境中使用时,元组可以避免数据被意外修改。

my_tuple = ('abcd', 786, 2.23, 'runoob', 70.2) # 避免使用 tuple 作为变量名
tinytuple = (123, 'runoob')
print(my_tuple) # 输出完整元组
print(my_tuple[0]) # 输出第一个元素:abcd
print(my_tuple[1:3]) # 输出索引 1 和 2 的元素:(786, 2.23)
print(my_tuple[2:]) # 输出从索引 2 开始的所有元素
print(tinytuple * 2) # 输出两次 tinytuple
print(my_tuple + tinytuple) # 连接两个元组
#warning 元组虽然是不可变的,但如果元组中包含可变对象(如列表),则这些可变对象仍然可以被修改。
my_tuple_with_list = (1, 2, [3, 4])
my_tuple_with_list[2][0] = 99 # 修改元组中的列表元素
#同时元组支持切片操作,类似于列表,但元组的切片操作会返回一个新的元组,而不是列表。
#元组没有append()、remove()等方法,因为它们是不可变的,但可以使用 + 运算符连接元组,或者使用 * 运算符重复元组。

使用元组的场景

  1. 不可变性:当你需要一个不可变的序列时,使用元组可以确保数据不会被意外修改。
  2. 哈希性:元组可以作为字典的键,而列表不行,因为元组是不可变的。
  3. 性能:元组通常比列表更节省内存和更快,因为它们是不可变的,Python可以对它们进行优化。 元组通常。比列表更节省内存和更快,因为它们是不可变的,Python可以对它们进行优化。 创建速度更快:创建元组比创建列表快。 内存占用更小:元组占用的内存空间比同等元素的列表少。 缓存机制:Python 内部会缓存一些常用的空元组或小元组,避免重复分配内存

2.2.4 集合#

Python 中的集合(Set)是一种无序、可变的数据类型,用于存储唯一的元素。 集合中的元素是唯一的,不能重复。 集合使用大括号 {} 定义,元素之间用逗号分隔。集合支持数学上的集合操作,如并集、交集、差集等。 类似于c++中的std::set和java中的HashSet,但python的集合更为灵活和强大。

sites = {'Google', 'Taobao', 'Runoob', 'Facebook', 'Zhihu', 'Baidu'} #使用的是字面量语法(大括号 {})。这是直接定义一个包含多个字符串元素的集合。
print(sites) # 输出集合(无序,重复元素会被自动去掉)
# 成员测试
if 'Runoob' in sites:
print('Runoob 在集合中')
else:
print('Runoob 不在集合中')
# set 可以进行集合运算
a = set('abracadabra') #set() 接收一个字符串时,它会将字符串打散,把里面的每一个字符当作独立的元素放入集合中。
b = set('alacazam')
print(a) # a 中的唯一字符
print(a - b) # a 和 b 的差集(在 a 中但不在 b 中)
print(a | b) # a 和 b 的并集(在 a 或 b 中)
print(a & b) # a 和 b 的交集(同时在 a 和 b 中)
print(a ^ b) # a 和 b 的对称差集(在 a 或 b 中,但不同时存在)
#常用操作
1.添加元素
sites.add('Twitter') # 添加元素
2.删除元素
sites.remove('Baidu') # 删除元素,如果元素不存在会抛出 KeyError 异常
3.清空集合
sites.clear() # 清空集合
4.集合长度
len(sites) # 获取集合的长度(元素个数)
5.集合遍历
for site in sites:
print(site) # 遍历集合中的元素
6.元素是否存在
if 'Google' in sites:
print('Google 在集合中')
else:
print('Google 不在集合中')

使用集合的场景:

  1. 去重:当你需要从一个列表或其他可迭代对象中去除重复元素时,集合是一个很好的选择。
  2. 数学运算:如果你需要进行集合的交集、并集、差集等操作,集合提供了方便的方法。
  3. 快速查找:集合的成员测试(in 操作)通常比列表更快,尤其是在处理大量数据时。

2.2.5 字典#

字典是一种映射类型,用 {} 标识,它是一个 键(key) : 值(value) 的集合。键(key) 必须使用不可变类型,且在同一个字典中键必须是唯一的。 注意: 1.Python 3.7 起,字典会保持元素的插入顺序,不再是无序的。如果需要有序字典的特性,直接使用普通 dict 即可。

2.字典的键与值可以包含任意类型的数据。

my_dict = {}
my_dict['one'] = "1 - 菜鸟教程"
my_dict[2] = "2 - 菜鸟工具"
tinydict = {'name': 'runoob', 'code': 1, 'site': 'www.runoob.com'}
print(my_dict['one']) # 输出键为 'one' 的值
print(my_dict[2]) # 输出键为 2 的值
print(tinydict) # 输出完整的字典
print(tinydict.keys()) # 输出所有键
print(tinydict.values()) # 输出所有值
#元组作为字典的键
my_dict_with_tuple = {('a', 'b'): 1, ('c', 'd'): 2} # 使用元组作为字典的键
print(my_dict_with_tuple[('a', 'b')]) # 输出键为 ('a', 'b') 的值:1
#元组常用操作
1.访问元素
my_dict_with_tuple[('a', 'b')] = 3 # 修改键为 ('a', 'b') 的值为 3
2.删除元素
del my_dict_with_tuple[('c', 'd')] # 删除键为 ('c', 'd') 的键值对
3.遍历字典
for key, value in my_dict_with_tuple.items():
print(f"键: {key}, 值: {value}") # 遍历字典中的键值对
4.字典长度
len(my_dict_with_tuple) # 获取字典的长度(键值对的数量)
5.检查键是否存在
if ('a', 'b') in my_dict_with_tuple.keys():
print("键 ('a', 'b') 存在")
else:
print("键 ('a', 'b') 不存在")
6.检查值是否存在
if 3 in my_dict_with_tuple.values():
print("值 3 存在")
else:
print("值 3 不存在")

2.2.6 bytes 与 bytearray#

作用:bytes 和 bytearray 都是用来表示二进制数据的类型,但它们有重要的区别。bytes 是不可变的,而 bytearray 是可变的。它们都用于处理二进制数据,例如文件读写、网络通信等场景。

特性bytesbytearray
可变性❌ 不可变✅ 可变
内存创建后固定可动态修改
适用场景常量、哈希键、网络传输需要频繁修改的二进制数据
1.常见 bytes 操作
# ========== 1. 创建 bytes ==========
b1 = b'hello' # 字面量
b2 = bytes('你好', 'utf-8') # 字符串编码
b3 = bytes([72, 101, 108]) # 从整数列表创建 → b'Hel'
b4 = bytes(5) # 5个零字节 → b'\x00\x00\x00\x00\x00'
# ========== 2. 索引与切片(与 str 类似)==========
b = b'Hello, World!'
b[0] # 72(返回整数,不是 bytes!)
b[0:5] # b'Hello'
b[-1] # 33('!' 的 ASCII 值)
# ========== 3. 常用方法 ==========
b = b' Hello, World! '
b.strip() # b'Hello, World!'
b.lower() # b' hello, world! '
b.split(b',') # [b' Hello', b' World! ']
b.replace(b'World', b'Python') # b' Hello, Python! '
b.startswith(b' He') # True
# ========== 4. 编码与解码 ==========
text = '你好世界'
encoded = text.encode('utf-8') # str → bytes
decoded = encoded.decode('utf-8') # bytes → str → '你好世界'
# ========== 5. 拼接与重复 ==========
b1 = b'Hello'
b2 = b' World'
b1 + b2 # b'Hello World'
b'Ha' * 3 # b'HaHaHa'
# ========== 6. 判断与查找 ==========
b = b'abcdef'
b'cd' in b # True
b.index(b'd') # 3
b.count(b'a') # 1
# ========== 7. 与整数互转 ==========
b = b'\x00\x01\x02\x03'
list(b) # [0, 1, 2, 3]
bytes([255, 128, 0]) # b'\xff\x80\x00'
# ========== 8. hex / fromhex ==========
b = b'\xde\xad\xbe\xef'
b.hex() # 'deadbeef'
bytes.fromhex('deadbeef') # b'\xde\xad\xbe\xef'
#bytearray 的可变操作
ba = bytearray(b'Hello')
2. ✅ 可变操作(bytes 不支持)
ba[0] = 72 # 按索引修改(赋值整数)
ba[1:3] = b'EL' # 切片赋值
ba.append(33) # 追加字节 → bytearray(b'HELLO!')
ba.extend(b' World') # 扩展
ba.insert(5, 44) # 插入
ba.pop() # 弹出最后一个字节
del ba[5] # 删除指定位置
# ⚠️ 注意:ba[i] = 整数,ba[i:j] = bytes/bytearray
ba[0] = ord('h') # ✅ 正确
# ba[0] = b'h' # ❌ TypeError!
3. 典型应用场景
# 📡 网络通信:读取 socket 数据
data = sock.recv(1024) # 返回 bytes
header = data[:4] # 解析头部
# 📁 文件读写(二进制模式)
with open('image.png', 'rb') as f:
content = f.read() # bytes
# 🔧 协议解析/修改(需要可变)
buf = bytearray(header + payload)
buf[2] = 0x01 # 修改标志位
buf[4:8] = length.to_bytes(4, 'big') # 更新长度字段
sock.sendall(buf)
特性PythonJavaC++
不可变二进制bytesbyte[] (约定俗成) / String (UTF-8)const uint8_t* / std::string
可变二进制bytearraybyte[] (原生可变)uint8_t[] / std::vector<uint8_t>
底层数据单元整数 (0-255)byte 类型 (有符号, -128~127)char / uint8_t (无符号, 0-255)
内存管理自动垃圾回收 (GC)自动垃圾回收 (GC)手动管理 (new/delete) 或 RAII (std::vector)
文件读写模式open('f', 'rb') 返回 bytesFileInputStream 读入 byte[]std::ifstream 配合 std::ios::binary
与文本的界限极其严格 (strbytes 必须通过 .encode()/.decode() 转换)相对模糊 (byte[]String 转换需指定字符集)完全没有界限 (一切皆内存字节,文本和二进制无类型区分)
典型应用场景网络协议解析、加密、文件处理网络 I/O、序列化、加密算法底层驱动、高性能网络服务器、游戏引擎、音视频处理

2.3数据类型转换#

2.3.1 隐式类型转换#

即自动类型转换,Python 会根据操作数的类型自动进行类型提升。例如,当一个整数与一个浮点数进行运算时,整数会被自动转换为浮点数。

2.3.2 显式类型转换#

常用的显式类型转换函数有:

  • int(x):将 x 转换为整数类型。
  • float(x):将 x 转换为浮点数类型。
  • str(x):将 x 转换为字符串类型。
  • list(x):将 x 转换为列表类型。
  • tuple(x):将 x 转换为元组类型。
  • set(x):将 x 转换为集合类型。
  • dict(x):将 x 转换为字典类型(x 必须是可迭代的键值对)。
  • bytes(x):将 x 转换为 bytes 类型。
  • bytearray(x):将 x 转换为 bytearray 类型。

下面的内容适合作为课程讲义,默认学生已经具备 C++、Java 基础,因此重点突出 Python 与 C++/Java 的区别,并给出最常用的写法和示例。


2.4 分支、循环、函数#

Python 的程序控制结构与 C++、Java 基本一致,都包括顺序、分支、循环三种结构。

但 Python 的语法更加简洁,最大的特点是使用缩进表示代码块,而不是 {}


2.4.1 分支语句#

(1)if 语句#

Python 使用 ifelifelse 实现条件判断。

语法格式:

if 条件:
语句块1
elif 条件:
语句块2
else:
语句块3

注意:

  • 条件后必须加 :
  • 使用缩进表示代码块(通常4个空格)
  • 不需要使用大括号 {}

(2)Python 特点#

Python 可以直接进行链式比较:

age = 20
if 18 <= age < 60:
print("成年人")

相比 C++:

if(age >=18 && age <60)

Python 写法更加自然、简洁。


2.4.2 循环语句#

Python 提供两种循环:

  • while
  • for

其中 for 循环使用频率远高于 while


(1)while 循环#

语法:

while 条件:
循环体

示例:

i = 1
while i <= 5:
print(i)
i += 1

(2)for 循环#

Python 的 for 与 C++ 不同。

C++:

for(int i=0;i<5;i++)

Python:

for i in range(5):
print(i)

Python 的 for 实际上是遍历一个可迭代对象#

(3)range() 函数#

range() 函数用于生成一个整数序列,常用于控制循环次数。

最常用的三种形式:

range(stop) #此处 stop 表示生成的整数序列的结束值(不包含 stop),默认从 0 开始,步长为 1。
for i in range(5):
print(i)

结果:

0 1 2 3 4

range(start, stop)
for i in range(2, 6):
print(i)

结果:

2 3 4 5

range(start, stop, step) #开始,截至,步数
for i in range(0, 10, 2):
print(i)

结果:

0 2 4 6 8

注意:当 step=-1时意味着倒序遍历,即从右向左遍历。


(4)遍历列表#

Python 最常见的写法:

names = ["Tom", "Alice", "Bob"]
for name in names:
print(name)

输出:

Tom
Alice
Bob

相比 C++ 使用下标遍历更加简洁。


(5)break 与 continue#

与 C++、Java 用法基本一致。

break:结束整个循环

for i in range(10):
if i == 5:
break
print(i)

输出:

0
1
2
3
4

continue:跳过本次循环

for i in range(5):
if i == 2:
continue
print(i)

输出:

0
1
3
4

2.4.3 可更改对象与不可更改对象#

在 Python 中,strtuple 和数字是不可更改的对象,而 listdict 等则是可以修改的对象。

不可变类型:变量赋值 a = 5 后再赋值 a = 10,这里实际是新生成一个 int 值对象 10,再让 a 指向它,而 5 被丢弃。这个过程不是改变 a 的值,而是让 a 指向了新对象。

可变类型:变量赋值 la = [1, 2, 3, 4] 后再赋值 la[2] = 5,则是将列表 la 的第三个元素值更改。列表本身没有改变,只是其内部的一部分值被修改了。

Python 函数的参数传递:

不可变类型:类似 C++ 的值传递,如整数、字符串、元组。如 fun(a),函数内部重新绑定 a 不会影响函数外部的对象。

可变类型:类似 C++ 的引用传递,如列表、字典。如 fun(la),在函数内部修改列表内容后,函数外部的 la 也会受到影响。

Python 中一切都是对象。严格来说,不能简单地说是值传递还是引用传递,更准确的说法是:传递不可变对象或可变对象。

2.4.4 函数定义与调用#

函数用于封装可重复使用的代码,提高程序的可读性和复用性。

Python 使用 def 定义函数。


(1)定义函数#

语法:

def 函数名(参数):
函数体

例如:

def hello():
print("Hello Python")

调用:

hello()

输出:

Hello Python

(2)带参数函数#

def add(a, b):
return a + b
result = add(3, 5)
print(result)

输出:

8

(3)默认参数#

Python 支持默认参数,这是 C++ 常用但 Java 不支持的特性。

def greet(name="Python"):
print("Hello,", name)
greet()
greet("Alice")

输出:

Hello, Python
Hello, Alice

(4)返回多个值#

Python 可以一次返回多个值,本质上返回的是元组(Tuple)。

def calculate(a, b):
return a + b, a - b
x, y = calculate(10, 3)
print(x)
print(y)

输出:

13
7

相比 C++,无需结构体或引用参数即可返回多个结果。


(5)关键字参数#

Python 可以指定参数名称,提高代码可读性。

def student(name, age):
print(name, age)
student(age=20, name="Tom")

输出:

Tom 20

参数顺序可以改变,只需保证参数名正确。


(6)可变参数(了解)#

当参数个数不确定时,可以使用 *args

def total(*args):
print(sum(args))
total(1, 2, 3)
total(10, 20, 30, 40)

输出:

6
100

小结#

内容Python 特点与 C++/Java 对比
if 分支使用缩进表示代码块不使用 {}
for 循环for...in 遍历对象不采用 for(;;) 形式
range()快速生成整数序列常用于控制循环次数
while与 C++ 基本一致语法更简洁
函数定义使用 def无需声明返回类型
默认参数支持Java 不支持
多返回值可直接返回多个值C++/Java 实现较复杂
关键字参数支持按参数名传参C++/Java 不支持
可变参数*args类似 C++ 可变参数,但更安全、更易用

2.5 lambda 表达式#

Python 使用 lambda 创建匿名函数。

[!重要] lambda 函数可以有任意数量的参数,但只能包含一个表达式,并返回该表达式的结果。

2.5.1 语法格式#

lambda arguments: expression
  • lambda 是 Python 的关键字,用于定义 lambda 函数。
  • arguments 是参数列表,可以包含零个或多个参数,需写在冒号(:)前。
  • expression 是用于计算并返回结果的表达式。

示例:

square = lambda x: x ** 2
print(square(5)) # 25

2.5.2 lambda 函数的特点#

  • 匿名:没有函数名称,可以赋值给变量或作为参数传递给其他函数。
  • 简洁:通常只包含一行代码,适合编写简单函数。
  • 常见用途:作为 map()filter()reduce() 等函数的参数,将简单的处理逻辑直接传入这些函数。

例如,使用 lambda 作为 map() (映射)的参数,对列表中的每个元素求平方:

numbers = [1, 2, 3, 4]
squares = list(map(lambda x: x ** 2, numbers)) # map() 函数会将 lambda 函数应用到 numbers 列表的每个元素上,并返回一个迭代器,最后使用 list() 将其转换为列表。
print(squares) # [1, 4, 9, 16]

使用 lambda 作为 filter()(过滤)的参数,只保留列表中的偶数:

numbers = [1, 2, 3, 4, 5, 6]
even_numbers = list(filter(lambda x: x % 2 == 0, numbers))
print(even_numbers) # [2, 4, 6]

使用 lambda 作为 reduce() (计算总和)的参数,计算列表中所有元素的和。使用前需要从 functools 导入 reduce

from functools import reduce
numbers = [1, 2, 3, 4]
total = reduce(lambda x, y: x + y, numbers)
print(total) # 10

2.5.3 lambda 函数的综合使用#

除了 map()filter()reduce()lambda 还常用于 sorted()key 参数中,根据指定规则排序。下面综合使用这些函数,先筛选出及格成绩,再按成绩从高到低排序,最后计算及格学生的总分:

from functools import reduce
students = [
{"name": "小明", "score": 85},
{"name": "小红", "score": 92},
{"name": "小刚", "score": 58},
{"name": "小丽", "score": 76},
]
# filter():筛选出及格的学生
#students是一个列表,filter()函数会将lambda函数应用到students列表的每个元素上,并返回一个迭代器,最后使用list()将其转换为列表。
#lambda student: student["score"] >= 60是一个匿名函数,接收一个学生字典作为参数,返回该学生的成绩是否大于等于60。
passed = list(filter(lambda student: student["score"] >= 60, students))
# sorted():按成绩从高到低排序
ranked = sorted(passed, key=lambda student: student["score"], reverse=True) #reverse=True表示降序排列
# map():提取排序后的学生姓名
names = list(map(lambda student: student["name"], ranked))
# reduce():计算及格学生的总分
total_score = reduce(
lambda total, student: total + student["score"],
passed,
0,
) # reduce()三个参数,第一个参数是函数,第二个参数是可迭代对象,第三个参数是初始值,这里初始值为0,表示从0开始累加。
print(names) # ['小红', '小明', '小丽']
print(total_score) # 253

2.5.4 推导式#

推导式可以用简洁的语法,根据一个可迭代对象快速创建新的序列。常见类型包括:

  • 列表(list)推导式
  • 字典(dict)推导式
  • 集合(set)推导式
  • 生成器(generator)表达式

列表推导式#

列表推导式的基本格式为:

[表达式 for 变量 in 可迭代对象]

也可以添加条件,只保留符合条件的元素:

[表达式 for 变量 in 可迭代对象 if 条件]

其中,表达式用于生成列表元素,for用于遍历可迭代对象,if用于过滤元素。

1. 生成指定范围的数字#

squares = [number ** 2 for number in range(1, 6)]
#其中,`number ** 2` 是表达式,表示将 `number` 的平方作为列表元素;`for number in range(1, 6)` 表示遍历从 1 到 5 的整数。
print(squares) # [1, 4, 9, 16, 25]

2. 过滤元素#

even_numbers = [number for number in range(10) if number % 2 == 0]
#其中,`number for number in range(10)` 表示遍历从 0 到 9 的整数;`if number % 2 == 0` 表示只保留偶数。
print(even_numbers) # [0, 2, 4, 6, 8]

3. 转换字符串#

names = ["alice", "bob", "carol"]
upper_names = [name.upper() for name in names]
print(upper_names) # ['ALICE', 'BOB', 'CAROL']

4. 使用条件表达式#

scores = [58, 72, 90]
results = ["及格" if score >= 60 else "不及格" for score in scores]
print(results) # ['不及格', '及格', '及格']

5. 字典推导式#

字典推导式使用 {键: 值 for 变量 in 可迭代对象} 创建字典:

numbers = [1, 2, 3]
squares = {number: number ** 2 for number in numbers}
print(squares) # {1: 1, 2: 4, 3: 9}

6. 集合推导式#

集合推导式使用 {表达式 for 变量 in 可迭代对象} 创建集合,并会自动去重:

words = ["Python", "python", "Java"]
lower_words = {word.lower() for word in words}
print(lower_words) # {'python', 'java'}

7. 生成器表达式#

生成器表达式使用圆括号,采用惰性计算,适合处理较大的数据:

numbers = (number ** 2 for number in range(1, 4))
print(list(numbers)) # [1, 4, 9]

2.6 python面向对象编程(OOP)#

首先来了解一下,相比于 C++ 和 Java,Python 的面向对象编程具有以下特点:

特性PythonJavaC++
类型系统动态类型静态类型静态类型
封装机制下划线约定(非强制)关键字强制(private等)关键字强制(含友元)
多态实现鸭子类型(天然动态)接口/抽象类显式虚函数(virtual)
继承方式多重继承(C3算法)单继承 + 多接口多重继承(含虚继承)
内存管理自动垃圾回收(GC)自动垃圾回收(GC)手动管理 / 智能指针
当前对象引用显式 self隐式 this隐式 this 指针
  • 一切皆对象:整数、字符串、函数和类本身都是对象。
  • 动态类型:定义变量时无需声明类型,类型在运行时确定。
  • 鸭子类型:只要对象支持所需的操作,就可以被使用,无需显式继承特定类。

2.6.1 类的定义与实例化#

使用 class 关键字定义类,构造方法 __init__ 用于初始化实例属性,实例方法的第一个参数通常为 self

注意:

1. 在Python中,self是实例方法的第一个参数,它代表了类的实例本身,而非类,相当于Java中的this关键字。 2. __init__是类的构造方法,用于初始化实例属性。 3. 在类的内部,使用 def 关键字来定义一个方法,与一般函数定义不同,类方法必须包含参数 self。

基本语法格式如下:

class 类名:
类属性 = 属性值
def __init__(self, 参数1, 参数2):
self.实例属性1 = 参数1
self.实例属性2 = 参数2
def 方法名(self, 参数):
# 方法体
return 结果
对象 = 类名(参数1, 参数2)
对象.方法名(参数)
class Person:
def __init__(self, name, age):
self.name = name
self.age = age
def introduce(self):
return f"我叫{self.name},今年{self.age}岁"
# 实例化对象并调用方法
person = Person("小明", 18)
print(person.name) # 小明
print(person.introduce()) # 我叫小明,今年18岁

类属性由所有实例共享,实例属性则属于具体对象:

class Dog:
species = "犬科" # 类属性
def __init__(self, name):
self.name = name # 实例属性
def bark(self):
print(f"{self.name}:汪汪!")
dog = Dog("旺财")
print(Dog.species) # 犬科
dog.bark() # 旺财:汪汪!

2.6.2 继承与多态#

继承可以让子类复用父类的属性和方法,并根据需要扩展或重写行为。多态表示同一个操作作用于不同对象时,可以产生不同的结果。

Python、Java 与 C++ 的继承和多态对比#

对比项PythonJavaC++
类型系统动态类型静态类型静态类型
类继承支持多继承类只支持单继承,可实现多个接口支持多继承
方法重写直接在子类中定义同名方法,可使用 super()使用 @Override,通常依赖接口或父类通常使用 virtualoverride
方法解析按 MRO(方法解析顺序)查找方法沿单一父类链查找受继承方式、虚函数和虚继承影响
多态基础鸭子类型:关注对象能做什么关注对象声明的父类或接口类型关注继承关系和虚函数
是否必须显式继承统一接口通常不需要通常需要通常需要

1. 多继承与 MRO#

一个 Python 类可以同时继承多个父类。当多个父类具有同名方法时,Python 会按照 MRO 决定调用顺序。可以通过 类名.mro() 查看该顺序。

经典的菱形继承示例如下:

class Animal:
def speak(self):
print("动物发出声音")
class Flyable(Animal):
def speak(self):
print("会飞的动物发出声音")
class Swimmable(Animal):
def speak(self):
print("会游泳的动物发出声音")
class Duck(Flyable, Swimmable):
pass
duck = Duck()
duck.speak() # 会飞的动物发出声音
print([cls.__name__ for cls in Duck.mro()])
# ['Duck', 'Flyable', 'Swimmable', 'Animal', 'object']

Duck 先继承 Flyable,因此会优先找到 Flyable.speak()。在实际项目中,多继承常与 super() 配合使用;参与协作的类应采用一致的方法签名,并继续调用 super(),让 MRO 链上的实现都有机会执行。

2. 鸭子类型与多态#

Python 的多态通常不要求对象继承同一个父类。只要对象提供了所需的方法,就可以被同一段代码使用,这就是“鸭子类型”。

class Dog:
def speak(self):
return "汪汪"
class Cat:
def speak(self):
return "喵喵"
class Robot:
def speak(self):
return "你好,我是机器人"
def make_it_speak(obj):
print(obj.speak())
for speaker in (Dog(), Cat(), Robot()):
make_it_speak(speaker)

DogCatRobot 没有共同的自定义父类,但它们都实现了 speak(),因此 make_it_speak() 可以统一处理它们。与 Java 接口或 C++ 虚函数相比,这种方式更灵活,但错误通常要到运行时才会暴露。

注意:Python确实没有传统意义上的多态机制,它靠的是运行时名字查找。但“多态”作为一种编程现象(同一接口,不同行为)在Python中依然存在,只是实现路径完全不同。

2.6.3 封装与私有属性#

封装是把数据和操作数据的方法组织在类中,并控制外部代码如何访问内部状态。Python 更强调约定和明确的接口,而不是强制禁止访问。

1. 属性的访问约定#

写法含义典型用途
name公开属性允许类内外直接访问
_name受保护属性(约定)提示外部代码和子类谨慎使用
__name触发名称改写避免子类意外覆盖,表达类内部使用的意图

以两个下划线开头、结尾没有两个下划线的属性(如 __balance)会被 Python 改写为 _类名__属性名。因此它不能通过 对象.__balance 直接访问,但这并不是绝对的安全或权限机制,不应依赖它保存敏感信息。

class BankAccount:
def __init__(self, owner, balance=0):
self.owner = owner
self.__balance = balance
def deposit(self, amount):
if amount <= 0:
raise ValueError("存款金额必须大于 0")
self.__balance += amount
def get_balance(self):
return self.__balance
def __check_password(self, password):
return password == "123456"
def withdraw(self, amount, password):
if not self.__check_password(password):
raise PermissionError("密码错误")
if amount > self.__balance:
raise ValueError("余额不足")
self.__balance -= amount
account = BankAccount("小明", 1000)
account.deposit(500)
account.withdraw(200, "123456")
print(account.get_balance()) # 1300
# print(account.__balance) # AttributeError
# account.__check_password("123456") # AttributeError

示例中的 __balance 是私有属性,__check_password() 是私有方法。类的内部通过 self.__balanceself.__check_password() 使用它们,外部则通过公开方法与对象交互。

2. 实例方法与 self#

类中使用 def 定义实例方法时,第一个参数用于接收当前实例,按照约定命名为 self。调用 account.deposit(500) 时,Python 会自动把 account 作为 self 传入。

self 不是关键字,理论上可以换成其他名称,但统一使用 self 能让代码更容易阅读和维护。

3. 专有方法#

以双下划线开头和结尾的方法称为专有方法(也常称“魔术方法”或 dunder 方法)。它们定义对象如何响应 Python 的内置操作,通常由解释器间接调用。

专有方法触发方式作用
__init__ClassName(...)初始化新创建的实例
__del__对象被回收时执行清理逻辑,但调用时机不确定,不适合管理关键资源
__repr__repr(obj)、交互式环境返回供开发者查看的对象表示
__setitem__obj[key] = value按键或索引赋值
__getitem__obj[key]按键或索引取值
__len__len(obj)返回对象长度
__call__obj(...)让实例可以像函数一样调用
__add__obj + other定义加法
__sub__obj - other定义减法
__mul__obj * other定义乘法
__truediv__obj / other定义真除法
__mod__obj % other定义取余运算
__pow__obj ** other定义乘方运算

Python 3 不再使用 __cmp__。比较操作应分别实现 __eq____lt____le____gt____ge__ 等方法。

下面的类通过专有方法表现得像一个内置容器:

class ScoreBook:
def __init__(self):
self.__scores = {}
def __setitem__(self, name, score):
if not 0 <= score <= 100:
raise ValueError("分数必须在 0 到 100 之间")
self.__scores[name] = score
def __getitem__(self, name):
return self.__scores[name]
def __len__(self):
return len(self.__scores)
def __repr__(self):
return f"ScoreBook({self.__scores!r})"
scores = ScoreBook()
scores["小明"] = 95
scores["小红"] = 88
print(scores["小明"]) # 95
print(len(scores)) # 2
print(scores) # ScoreBook({'小明': 95, '小红': 88})

3. Python 高级进阶#

这一章围绕一个常见需求展开:创建独立开发环境,编写采集或处理数据的脚本,让程序运行得更稳定,再把结果保存下来。每节的完整 Python 示例都可单独运行;需要其他文件或第三方依赖时会提前说明。

示例使用 Python 3.11+,Windows 命令默认在 PowerShell 中执行,Linux/macOS 的差异单独列出。先通过 python --version 确认版本;安装依赖前,应激活自己项目的环境。文中的安装和联网命令供读者按需执行。

3.1 Python 虚拟环境#

3.1.1 为什么每个项目需要独立环境#

假设旧项目依赖某个库的 1.x 版本,新项目需要 2.x。如果全部安装到同一个 Python 环境,升级新项目可能使旧项目无法启动。虚拟环境为每个项目提供独立的依赖目录,让安装、升级、重建都能围绕项目进行。

全局安装导致项目依赖版本冲突,独立虚拟环境允许不同项目使用不同版本

图中的 Django 版本仅用于解释冲突,并非新项目的版本推荐。“隔离”主要指 Python 依赖;虚拟环境不隔离文件访问、网络、操作系统或 GPU 驱动,也不是运行不可信代码的安全沙箱。

3.1.2 venv 与 Conda 如何选择#

对比项venvConda
来源Python 标准库自带需要安装 Conda 发行版
Python 版本基于创建环境时已有的解释器可以在环境中安装指定版本解释器
管理范围Python 环境,通常搭配 pip 安装包Python 包以及渠道提供的非 Python 库、工具
典型场景Web 服务、自动化脚本、普通 Python 项目科研、数据分析、依赖复杂二进制库的项目
共享方式依赖清单或项目锁文件environment.yml,必要时加平台专用锁定信息
注意事项不要把 .venv 复制到另一台机器渠道、操作系统和驱动仍影响可复现性

普通脚本从 venv + pip 入门即可;团队已有 Conda 环境文件时,优先沿用。Conda 能安装部分 CUDA 运行库,但不能代替宿主机显卡驱动,深度学习环境仍需核对框架与驱动兼容性。

3.1.3 创建、激活和确认 venv#

在项目目录执行:

Terminal window
python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -c "import sys; print(sys.executable)"
python -m pip --version

Linux/macOS 的创建与激活命令如下:

python3 -m venv .venv
source .venv/bin/activate
python -c "import sys; print(sys.executable)"

输出的解释器路径应位于项目的 .venv 中。工作结束后执行 deactivate 退出;把 .venv/ 加入 .gitignore,提交源代码和依赖清单即可。

如果 PowerShell 阻止激活脚本,可以直接调用环境中的解释器,无需修改执行策略:

Terminal window
.\.venv\Scripts\python.exe -m pip --version
.\.venv\Scripts\python.exe main.py

第二条命令要求项目中已有 main.py。在 VS Code 中通过 Python: Select Interpreter 选择同一解释器,再新建终端。判断环境是否正确应看 sys.executable,不能只看终端前面的环境名。venv 官方说明

3.1.4 Conda 常用工作流#

下面假设 Conda 已安装且当前终端可以执行 conda

Terminal window
conda create -n data-lab python=3.12
conda activate data-lab
conda install numpy pandas
conda env list
python -c "import sys; print(sys.executable)"
conda env export --from-history > environment.yml
conda deactivate

另一台机器使用 conda env create -f environment.yml 重建环境。--from-history 主要记录主动安装的 Conda 依赖,便于跨平台调整,但不锁定全部传递依赖,也不完整记录 pip 安装项;需要精确复现实验时,还应记录完整版本、平台和 pip 依赖。

优先安装 Conda 包,再在已激活的环境中用 python -m pip 补充渠道中没有的包。混用后避免反复让两个工具替换同一组依赖,环境冲突严重时从清单重建通常更清晰。Conda 环境管理

3.2 Python 的包管理工具#

3.2.1 环境、依赖声明和锁定的区别#

工具或文件解决的问题实践建议
pip给指定 Python 安装、卸载发行包使用 python -m pip 避免装到其他解释器
requirements.txt批量安装一组依赖可以指定版本,也可以引用带哈希的固定依赖清单
pyproject.toml声明项目元数据、依赖和工具配置用于可维护的项目,不只是一次性脚本
uv环境、依赖解析、锁定与项目运行团队采用后统一使用其工作流
Poetry项目依赖、锁定、构建和发布已有 Poetry 项目继续使用原有配置
pipx为独立命令行工具提供隔离环境适合安装工具,不负责项目运行依赖

依赖声明描述“允许使用哪些版本”,锁文件记录解析出的具体依赖。锁文件仍不能保证不同操作系统、CPU 架构和外部服务完全一致。

3.2.2 pip 的日常命令#

以下操作在虚拟环境中执行:

Terminal window
python -m pip install requests
python -m pip show requests
python -m pip list
python -m pip check
python -m pip freeze > requirements.txt

重建环境时使用 python -m pip install -r requirements.txt;主动升级用 python -m pip install --upgrade requests,卸载用 python -m pip uninstall requests。升级后应运行相关测试,再更新依赖清单。

pip freeze 输出当前环境中已安装包的清单,不会为项目生成完整的跨平台锁定方案;请在干净的项目环境中导出。Windows PowerShell 5.1 的 > 可能写成 UTF-16,可改为:

Terminal window
python -m pip freeze | Out-File -Encoding utf8 requirements.txt

包的发行名称与导入名称也可能不同,例如安装 beautifulsoup4,代码中却是 from bs4 import BeautifulSoup。遇到 ModuleNotFoundError,先比对解释器路径与 python -m pip show 包名 的安装位置。pip 用户指南

3.2.3 一个 uv 项目工作流#

以下命令要求已安装 uv,并且在准备存放项目的父目录执行:

Terminal window
uv init collector-demo
cd collector-demo
uv add requests
uv add --dev pytest
uv run python -c "import requests; print(requests.__version__)"
uv sync --locked

uv add 更新项目依赖与锁文件;uv run 在项目环境中执行命令;uv sync --locked 在锁文件与声明不一致时失败,适合 CI 检查。提交 pyproject.tomluv.lock,让协作者使用同一套依赖解析结果。uv 项目指南

3.2.4 安装失败时检查什么#

现象优先检查
已安装却无法导入当前解释器、虚拟环境、发行名与导入名
No matching distribution foundPython 版本、平台、架构、包名和配置的索引
构建 wheel 失败是否缺少编译器或系统库,是否有适配的预编译包
依赖版本冲突阅读冲突链,调整直接依赖约束,而非强行忽略依赖
下载超时网络、代理、组织指定的软件源

排障时保留最早的有效错误信息。不要为了安装成功随意关闭证书校验,也不要把包含令牌的私有源地址提交到仓库。

3.3 Python 的性能优化#

3.3.1 先定位慢在哪里#

处理一份大日志时,瓶颈可能是磁盘、字符串解析、重复查找,也可能是网络等待。先记录输入规模、Python 版本、总耗时和峰值内存,再决定优化方法。

工具适合测量注意事项
time.perf_counter()整段业务耗时包含等待时间,适合比较端到端体验
timeit小段纯计算代码重复测量,避免把初始化混入比较
cProfile函数调用次数和累计耗时找热点函数,而非凭感觉改代码
tracemallocPython 内存分配不等于整个进程内存,也不覆盖所有原生库分配

如果已有 report.py,可执行:

Terminal window
python -m cProfile -s cumulative report.py

先看累计耗时高且调用频繁的函数。把计算移到另一行、把循环改成推导式,都不一定改变真正的瓶颈。

3.3.2 实战:用集合过滤重复编号#

场景:从一批编号中筛选未处理的数据。列表成员查找通常是 O(n),集合成员查找平均为 O(1),大批量判断时差异很明显。

from timeit import repeat
processed = list(range(10_000))
incoming = list(range(9_000, 11_000))
def filter_with_list():
return [item for item in incoming if item not in processed]
def filter_with_set():
processed_set = set(processed)
return [item for item in incoming if item not in processed_set]
assert filter_with_list() == filter_with_set()
for func in (filter_with_list, filter_with_set):
samples = repeat(func, number=3, repeat=3)
print(func.__name__, f"{min(samples) / 3:.6f} 秒/次")

此处将建集合的成本也计入测量。若同一个集合能用于多批任务,可把构建移到批次外。集合会占用额外内存,元素必须可哈希;它适合判断是否存在,不适合保留重复次数。输出耗时随硬件变化,应比较相同环境下的结果。

3.3.3 实战:逐行处理大文件#

不需要一次性拿到全部内容时,用迭代器避免 read() 后再 splitlines() 创建大对象:

from pathlib import Path
from tempfile import TemporaryDirectory
def iter_errors(path):
with path.open(encoding="utf-8") as file:
for number, line in enumerate(file, start=1):
if "ERROR" in line:
yield number, line.rstrip("\n")
with TemporaryDirectory() as directory:
path = Path(directory) / "app.log"
path.write_text("INFO start\nERROR timeout\nINFO done\n", encoding="utf-8")
for number, message in iter_errors(path):
print(number, message) # 2 ERROR timeout

yield 让调用方按需取数据。内存主要取决于当前行和下游是否累积结果;如果立刻调用 list(iter_errors(path)),仍会保存所有匹配项。

3.3.4 缓存、批处理与优化边界#

重复执行昂贵的纯函数时,可以考虑 functools.lru_cache(maxsize=...);参数需可哈希,结果不能依赖随时变化的外部状态。数据库查询和 HTTP 请求更适合先减少次数、批量处理、复用连接,再考虑增加并发。

优化后至少确认结果相同,并比较代表性小输入和大输入。不要给未设失效策略的缓存无限增长空间,也不要假设换成线程就能加速所有 Python 计算。

3.4 Python 的多线程与异步编程#

3.4.1 根据任务选择并发模型#

按任务瓶颈选择线程、异步或进程:阻塞 I/O 用线程,异步 I/O 用 asyncio,纯 Python CPU 计算考虑进程

工作类型常用方案典型场景
阻塞 I/O,已有同步库ThreadPoolExecutor多文件读取、同步 HTTP 客户端
大量等待,使用异步库asyncio多个 API 请求、长连接服务
纯 Python CPU 密集计算ProcessPoolExecutor独立的文本分析、计算任务
小任务、数量很少顺序执行并发开销可能超过收益

常规启用 GIL 的 CPython 中,同一进程通常不能由多个线程同时执行 Python 字节码,但 I/O 等待时可释放 GIL。某些原生扩展也会释放 GIL。Python 3.13 起提供可选的 free-threaded 构建,不能据此假定日常安装的解释器已关闭 GIL;应根据实际构建和依赖测量。线程与 GIL 说明

3.4.2 线程池:并发执行阻塞任务#

下面用短暂休眠模拟三个耗时不确定的 I/O 任务,不需要外网。其中一个任务故意失败,演示如何保留其他任务的结果。

from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep
def read_record(record_id):
sleep(0.02)
if record_id == 2:
raise ValueError("记录格式无效")
return {"id": record_id, "status": "ok"}
with ThreadPoolExecutor(max_workers=3) as pool:
futures = {pool.submit(read_record, number): number for number in range(3)}
for future in as_completed(futures):
record_id = futures[future]
try:
print("成功", future.result())
except ValueError as exc:
print("失败", record_id, str(exc))

结果的打印顺序不保证一致。限制 max_workers 可以控制同时运行的任务数,但一次提交百万个任务仍会占用大量内存,应分批提交或使用有界队列。

尽量让工作线程返回结果,由主线程合并。多个线程共同修改计数器等共享状态时,要使用 threading.Lock 或线程安全队列;不能依靠 GIL 保证一整段业务操作的原子性。

3.4.3 asyncio:限并发、超时与结果汇总#

async def 定义协程函数;调用它只会产生协程对象,需要 await 或交给事件循环执行。下面模拟四个请求,同时最多执行两个,其中一个超时:

import asyncio
async def fetch_record(record_id, semaphore):
async with semaphore:
try:
async with asyncio.timeout(0.08):
delay = 0.15 if record_id == 3 else 0.01
await asyncio.sleep(delay)
return {"id": record_id, "status": "ok"}
except TimeoutError:
return {"id": record_id, "status": "timeout"}
async def main():
semaphore = asyncio.Semaphore(2)
async with asyncio.TaskGroup() as group:
tasks = [
group.create_task(fetch_record(number, semaphore))
for number in range(4)
]
print([task.result() for task in tasks])
if __name__ == "__main__":
asyncio.run(main())

结果按 tasks 列表顺序汇总:编号 0、1、2 为 ok,编号 3 为 timeout。这里的超时从拿到信号量后开始,不包括排队等待。TaskGroup 遇到未处理的普通异常时会取消其他任务,并在退出时报告异常组;示例把预期超时转换成结果,以便整批任务继续。asyncio 任务文档

在协程中直接执行 time.sleep() 或同步网络请求会阻塞事件循环。可使用对应异步库,或用 await asyncio.to_thread(blocking_function, ...) 迁移阻塞 I/O;取消等待不一定能停止后台线程,所以底层网络请求仍需自己的超时。

3.4.4 CPU 计算:进程池的入口保护#

下面保存为 process_demo.py,使用 python process_demo.py 运行:

from concurrent.futures import ProcessPoolExecutor
def sum_squares(limit):
return sum(number * number for number in range(limit))
if __name__ == "__main__":
with ProcessPoolExecutor(max_workers=2) as pool:
results = list(pool.map(sum_squares, [10_000, 20_000]))
print(results) # [333283335000, 2666466670000]

工作函数定义在模块顶层,参数与返回值需要可序列化。入口保护尤其关系到 Windows 的子进程启动;不要把这个例子直接粘进交互式控制台。这里只演示结构,小规模计算未必比单进程快。

3.5 Python 的装饰器#

3.5.1 使用场景与执行顺序#

多个函数都要统计耗时、检查权限或记录调用时,可以把公共行为写成装饰器。装饰器接收函数,返回包装后的可调用对象。

@decorator
def work():
...

上面是语法示意,等价于先定义 work,再执行 work = decorator(work)。装饰通常发生在函数定义时,包装函数的函数体则在每次调用时执行。

3.5.2 实战:保留函数信息的计时装饰器#

from functools import wraps
from time import perf_counter
def timed(func):
@wraps(func)
def wrapper(*args, **kwargs):
started = perf_counter()
try:
return func(*args, **kwargs)
finally:
elapsed = perf_counter() - started
print(f"{func.__name__}: {elapsed:.6f}s")
return wrapper
@timed
def total_price(prices, discount=1.0):
"""计算折扣后的总价。"""
return sum(prices) * discount
print(total_price([10, 20], discount=0.9)) # 先输出耗时,再输出 27.0
print(total_price.__name__) # total_price

*args**kwargs 转发参数,return 保留原函数返回值,finally 保证异常时也记录耗时,wraps 保留名称、文档和 __wrapped__ 等元数据。原函数异常仍会抛出,调用者可以正常处理。

3.5.3 实战:带参数的有限重试#

重试适合短暂网络故障等可恢复错误,不能重试所有异常。下面只演示同步函数,且用本地模拟操作避免真实网络波动:

from functools import wraps
from time import sleep
def retry(*, attempts=3, delay=0.1, exceptions=(TimeoutError,)):
if attempts < 1 or delay < 0:
raise ValueError("attempts 至少为 1,delay 不能为负数")
def decorate(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(attempts):
try:
return func(*args, **kwargs)
except exceptions:
if attempt == attempts - 1:
raise
sleep(delay * (2 ** attempt))
return wrapper
return decorate
def make_demo_reader():
calls = 0
@retry(attempts=3, delay=0)
def read():
nonlocal calls
calls += 1
if calls < 3:
raise TimeoutError("暂时不可用")
return "第三次读取成功"
return read
print(make_demo_reader()()) # 第三次读取成功

生产环境还需总时限、抖动和服务端限流处理。对于扣款、创建订单等有副作用的请求,先设计幂等键或去重机制,避免重试造成重复操作。

@timed 放在 @retry(...) 上面时,计时覆盖整个重试过程;顺序颠倒时,每次尝试单独计时。装饰器从下往上应用。异步函数需要 async def wrapperawait func(...),否则统计到的可能只是创建协程对象的时间。

3.6 Python 的网络编程#

3.6.1 日常开发优先理解 HTTP#

调用开放 API 时,通常使用 HTTP 客户端,而不是直接手写 TCP 协议。一次请求包含方法、URL、请求头和可选请求体;响应包含状态码、响应头和响应体。

情况含义处理方式
2xx请求被成功处理继续检查响应格式和业务字段
4xx请求、权限或资源问题检查参数和授权,不盲目重试
429触发限流遵守 Retry-After,降低请求频率
5xx服务端处理失败对适合重试的操作有限重试
超时、连接失败可能没有收到有效 HTTP 响应记录上下文,设置明确失败路径

3.6.2 实战:本地 JSON 服务与客户端#

先把以下代码保存为 demo_server.py。它只监听本机回环地址,供后续章节练习;不用于生产部署。

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlsplit
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlsplit(self.path)
if parsed.path != "/api/hello":
self.send_error(404, "Not Found")
return
name = parse_qs(parsed.query).get("name", ["访客"])[0]
body = json.dumps({"message": f"你好,{name}"}, ensure_ascii=False).encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
if __name__ == "__main__":
with ThreadingHTTPServer(("127.0.0.1", 8765), Handler) as server:
print("本地服务:http://127.0.0.1:8765,Ctrl+C 退出")
try:
server.serve_forever()
except KeyboardInterrupt:
pass

在一个终端运行 python demo_server.py,再在另一个终端运行下面的 demo_client.py。如果 8765 已被占用,先选择空闲端口,并同步修改两端代码。

import json
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
query = urlencode({"name": "小明"})
request = Request(
f"http://127.0.0.1:8765/api/hello?{query}",
headers={"Accept": "application/json"},
)
try:
with urlopen(request, timeout=3) as response:
body = response.read(1_000_001)
if len(body) > 1_000_000:
raise ValueError("响应超过示例允许的大小")
data = json.loads(body.decode("utf-8"))
print(data["message"]) # 你好,小明
except HTTPError as exc:
print("HTTP 状态错误:", exc.code)
except (URLError, TimeoutError) as exc:
print("连接失败或超时:", exc)
except (ValueError, KeyError, TypeError) as exc:
print("响应内容不符合预期:", exc)

urlencode() 负责查询参数转义;不要直接把包含空格、中文或 & 的值拼到 URL 中。响应通过上下文管理器关闭,读取设置了大小上限。标准库的 timeout 主要约束阻塞操作,不是整个业务流程的绝对总时限。

3.6.3 第三方客户端与连接复用#

接口调用较多时,Requests 提供更简洁的参数与会话管理。先在当前环境执行 python -m pip install requests,并保持前面的本地服务运行:

import requests
try:
with requests.Session() as session:
response = session.get(
"http://127.0.0.1:8765/api/hello",
params={"name": "小明"},
timeout=(3, 5),
)
response.raise_for_status()
print(response.json()["message"])
except requests.RequestException as exc:
print("请求失败:", exc)
except (ValueError, KeyError, TypeError) as exc:
print("响应格式错误:", exc)

timeout=(3, 5) 分别设置连接和读取超时,不等于请求总耗时最多 8 秒。Session 有助于复用连接。真实 HTTPS 请求保留默认的证书校验;凭据从环境变量或凭据系统读取,日志中不要打印授权头。Requests 使用说明

3.6.4 TCP 与 HTTP 的边界#

TCP 提供有序字节流,没有消息边界:一次 send() 不一定对应一次 recv()recv(1024) 也不保证收满 1024 字节。自定义协议需要长度前缀或分隔符、接收缓冲区、超时和消息大小限制。对接已有 HTTP API 时,把这些底层细节交给成熟客户端通常更合适。

3.7 JSON 与 Python 的交互#

3.7.1 类型映射和四个常用函数#

JSONPython需要注意
objectdictJSON 对象的键是字符串
arraylistPython 元组序列化后再读取会变成列表
stringstrJSON 字符串使用双引号
numberint / float金额不要依赖二进制浮点精确计算
true / falseTrue / False大小写不同
nullNone不等于空字符串或缺失字段

dumps() / loads() 在对象与字符串间转换,dump() / load() 面向文件对象。JSON 不支持注释、尾随逗号和 Python 的单引号字典写法。

import json
record = {"name": "小明", "active": True, "tags": ["Python", "数据处理"]}
text = json.dumps(record, ensure_ascii=False, indent=2, allow_nan=False)
restored = json.loads(text)
print(text)
assert restored == record

ensure_ascii=False 保留可读中文,不决定文件编码;写文件时仍需指定 encoding="utf-8"allow_nan=False 防止生成标准 JSON 不允许的 NaNInfinity 等值。

3.7.2 实战:校验 API 返回的任务数据#

“能解析成 JSON”只说明语法有效,不代表数据符合业务要求。下面的函数接收 JSON 字符串并检查必要字段:

import json
def parse_task(text):
try:
data = json.loads(text)
except json.JSONDecodeError as exc:
raise ValueError(f"JSON 语法错误,第 {exc.lineno} 行") from exc
if not isinstance(data, dict):
raise ValueError("任务必须是对象")
if type(data.get("id")) is not int or data["id"] <= 0:
raise ValueError("id 必须是正整数")
title = data.get("title")
if not isinstance(title, str) or not title.strip():
raise ValueError("title 必须是非空字符串")
return {"id": data["id"], "title": title.strip()}
print(parse_task('{"id": 1, "title": " 整理日志 "}'))
try:
parse_task('{"id": true, "title": "错误示例"}')
except ValueError as exc:
print(exc) # id 必须是正整数

这里刻意使用 type(...) is int,因为 Python 中 boolint 的子类,而任务编号不应接受布尔值。不要使用 eval() 解析外部 JSON,它会执行 Python 表达式。

3.7.3 日期、金额和大批量记录#

JSON 没有日期或 Decimal 类型。应约定明确的传输格式,例如带时区的 ISO 8601 字符串,以及以分计的整数金额:

import json
from datetime import datetime, timezone
order = {
"created_at": datetime(2026, 1, 1, tzinfo=timezone.utc).isoformat(),
"amount_cents": 1990,
"currency": "CNY",
}
print(json.dumps(order, ensure_ascii=False, allow_nan=False))

不建议用 default=str 随意转换所有未知对象,这容易把类型错误隐藏成字符串。金额若采用字符串形式,也要约定币种和小数位规则。

采集数据可用 JSON Lines(JSONL):每行一个完整 JSON 对象,便于追加和流式读取。不要连续向同一普通 JSON 文件调用多次 json.dump(),否则会产生相邻对象而非合法的单一 JSON 文档。

3.8 Python 的文件操作与数据持久化#

3.8.1 先选择存储形式#

需求常用形式局限
配置、少量嵌套记录JSON整体修改通常需要重写
规则表格、交给表格软件CSV类型和嵌套结构表达有限
逐条日志、采集结果JSONL多进程同时追加需要协调
本地查询、更新、事务SQLite同时写入的并发能力有限
Python 对象临时缓存pickle不可加载不可信数据,不适合作为公共交换格式

3.8.2 路径、编码与文件模式#

使用 pathlib.Path 组合路径。相对路径相对于进程工作目录,而不是自动相对于脚本;脚本旁文件可通过 Path(__file__).resolve().parent 定位,Notebook 中通常没有 __file__

模式行为
r读取,文件不存在时报错
w创建或清空后写入
a末尾追加
x仅创建新文件,已存在则报错
rb / wb读写二进制数据,不指定文本编码

下面用临时目录演示 JSONL,运行结束会清理示例数据:

import json
from pathlib import Path
from tempfile import TemporaryDirectory
with TemporaryDirectory() as directory:
path = Path(directory) / "records.jsonl"
with path.open("w", encoding="utf-8") as file:
for number in range(3):
file.write(json.dumps({"id": number}, ensure_ascii=False) + "\n")
with path.open(encoding="utf-8") as file:
for number, line in enumerate(file, start=1):
try:
print(json.loads(line))
except json.JSONDecodeError as exc:
raise ValueError(f"第 {number} 行 JSON 无效") from exc

写 CSV 时使用 newline=""csv 模块处理换行;JSON 中的中文则同时需要正确的文件编码。文件句柄用 with 管理,异常时也能关闭。

3.8.3 实战:替换配置文件,避免留下半份 JSON#

直接用 w 打开已有配置会先清空旧内容。如果写入中断,配置可能损坏。可以先在同目录写临时文件,成功后再替换目标:

import json
import os
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory
def save_json(path, data):
path = Path(path)
text = json.dumps(data, ensure_ascii=False, indent=2, allow_nan=False)
path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = None
try:
with NamedTemporaryFile(
mode="w", encoding="utf-8", dir=path.parent, delete=False
) as file:
temporary_path = Path(file.name)
file.write(text)
file.flush()
os.fsync(file.fileno())
os.replace(temporary_path, path)
finally:
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
with TemporaryDirectory() as directory:
path = Path(directory) / "settings.json"
save_json(path, {"theme": "dark", "language": "zh-CN"})
print(json.loads(path.read_text(encoding="utf-8")))

临时文件放在同目录以避免跨文件系统移动;先关闭文件也兼容 Windows。替换可以减少读取到半份文件的风险,但不等于多写入者事务,也不保证所有文件系统在掉电后的持久性。多个进程更新同一配置时仍需锁或数据库。

3.8.4 实战:用 SQLite 保存和更新采集结果#

标准库 sqlite3 无需另装数据库服务,适合单机工具。示例数据库保存在临时目录;实际项目换成自己的数据路径即可。

import sqlite3
from contextlib import closing
from pathlib import Path
from tempfile import TemporaryDirectory
with TemporaryDirectory() as directory:
path = Path(directory) / "articles.db"
with closing(sqlite3.connect(path)) as connection:
connection.row_factory = sqlite3.Row
with connection:
connection.execute(
"CREATE TABLE articles (url TEXT PRIMARY KEY, title TEXT NOT NULL)"
)
connection.executemany(
"INSERT INTO articles (url, title) VALUES (?, ?) "
"ON CONFLICT(url) DO UPDATE SET title = excluded.title",
[
("https://example.com/1", "第一篇"),
("https://example.com/2", "第二篇"),
("https://example.com/1", "第一篇(更新)"),
],
)
rows = connection.execute("SELECT url, title FROM articles ORDER BY url")
print([dict(row) for row in rows])

? 占位符绑定数据,避免把外部值拼接进 SQL;它不能用于替换表名等 SQL 标识符。主键和 UPSERT 让重复采集更新已有记录。内层 with connection 成功时提交、异常时回滚,外层 closing 才负责关闭连接。sqlite3 连接上下文说明

并发程序最好把数据交给专门的写入线程或进程;不要随意跨线程共享连接。出现 database is locked 时先缩短事务、减少同时写入,再考虑更适合多用户服务的数据库。

3.9 Python 爬虫与 Scrapy 库#

3.9.1 从任务规模选择工具#

任务优先考虑
网站有正式数据 API直接调用 API,校验结构化结果
少量静态页面HTTP 客户端加 HTML 解析库
多页面、分页、清洗、导出Scrapy
内容必须执行 JavaScript 后才出现先确认允许的接口,再评估浏览器自动化

先确认目标站点允许的访问方式、服务条款及 robots 规则,再限制频率和范围。Scrapy 默认下载 HTML,不会像浏览器一样自动执行页面中的 JavaScript。

Scrapy 从请求队列下载网页,Spider 解析后产生后续请求或数据,数据经过 Pipeline 清洗再导出

3.9.2 建立可运行的练习项目#

以下使用 Scrapy 官方教程中的练习站点 https://quotes.toscrape.com/。需要可访问外网,先在独立环境中安装 Scrapy:

Terminal window
python -m pip install scrapy
scrapy startproject quote_lab
cd quote_lab

生成目录的主要职责如下:

quote_lab/
scrapy.cfg
quote_lab/
settings.py # 下载、并发与导出配置
items.py # 可选的数据字段定义
pipelines.py # 清洗和持久化
spiders/
__init__.py
quotes.py # 接下来新建的爬虫

3.9.3 提取内容并跟随分页#

quote_lab/spiders/quotes.py 写入以下完整 Spider:

import scrapy
class QuotesSpider(scrapy.Spider):
name = "quotes"
allowed_domains = ["quotes.toscrape.com"]
start_urls = ["https://quotes.toscrape.com/"]
def parse(self, response):
for quote in response.css("div.quote"):
yield {
"text": quote.css("span.text::text").get(),
"author": quote.css("small.author::text").get(),
"tags": quote.css("div.tags a.tag::text").getall(),
"source": response.url,
}
next_page = response.css("li.next a::attr(href)").get()
if next_page:
yield response.follow(next_page, callback=self.parse)

.get() 返回首个匹配或 None.getall() 返回列表;response.follow() 根据当前页面解析相对链接。yield dict 产生数据,yield Request 安排后续请求,网络下载由 Scrapy 调度。Scrapy 官方教程

3.9.4 设置超时、限速和采集上限#

在生成的 quote_lab/settings.py 中编辑或加入以下配置,避免保留同名配置的多个有效定义:

ROBOTSTXT_OBEY = True
CONCURRENT_REQUESTS = 4
CONCURRENT_REQUESTS_PER_DOMAIN = 2
DOWNLOAD_DELAY = 1.0
DOWNLOAD_TIMEOUT = 15
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0
RETRY_TIMES = 2
DEPTH_LIMIT = 3
FEED_EXPORT_ENCODING = "utf-8"

DEPTH_LIMIT=3 限制请求链深度,用来缩小练习范围;它不是精确的页面数量限制。DOWNLOAD_DELAY 与自动限速共同约束抓取,实际请求间隔会变化。重试、并发与延迟应按站点承载能力设置,不能把遭到限流当作继续增大并发的理由。Scrapy 配置参考

3.9.5 Pipeline:校验、清洗和去重#

把下面的类加入 quote_lab/pipelines.py

from scrapy.exceptions import DropItem
class CleanQuotePipeline:
def __init__(self):
self.seen = set()
def process_item(self, item, spider):
text = item.get("text")
author = item.get("author")
if not isinstance(text, str) or not isinstance(author, str):
raise DropItem("缺少正文或作者")
text, author = text.strip(), author.strip()
if not text or not author:
raise DropItem("正文或作者为空")
key = (text, author)
if key in self.seen:
raise DropItem("重复记录")
self.seen.add(key)
item["text"], item["author"] = key
return item

再在 settings.py 中注册:

ITEM_PIPELINES = {"quote_lab.pipelines.CleanQuotePipeline": 300}

数字越小越早执行;管道应返回清洗后的条目或抛出 DropItem。这里的集合只在本次进程中去重,大规模或跨次运行去重应改用数据库唯一约束,避免内存不断增长。

3.9.6 运行、导出和定位问题#

在有 scrapy.cfg 的项目根目录执行:

Terminal window
scrapy list
scrapy crawl quotes -O quotes.jsonl

-O 覆盖同名输出文件,重新运行前确认不需要保留旧结果;-o 对支持追加的输出格式通常追加,反复运行可能产生重复数据。这里选择 JSONL,便于逐条检查。

执行完后,用以下脚本检查前两条并统计数量:

import json
count = 0
with open("quotes.jsonl", encoding="utf-8") as file:
for line in file:
item = json.loads(line)
if not item.get("text") or not item.get("author"):
raise ValueError("输出存在无效条目")
count += 1
if count <= 2:
print(item["author"], item["text"])
print("有效条目数:", count)
问题排查方向
找不到 Spider执行目录、name、模块导入错误、当前环境
返回 200 却没有数据选择器是否匹配实际 HTML,是否只有 JS 渲染后才有内容
403 / 429 或 robots 拒绝权限、站点规则、访问频率;停止或调整合法访问方式
字段大量为 None检查页面结构,区分缺失字段与选择器错误
下一页没有抓取相对链接、域名限制、深度限制、重复请求过滤

可以使用 scrapy shell https://quotes.toscrape.com/ 单独验证选择器,再运行整个爬虫。查看日志中的响应状态、item_scraped_count、丢弃原因和结束原因,不能只凭程序退出判断采集成功。

完成这个练习后,可将 JSONL 清洗结果接入 3.8 节的 SQLite,把唯一键设为稳定的业务标识,补充采集时间与来源,并对重跑、空页面和网络失败进行验证。

分享

如果这篇文章对你有帮助,欢迎分享给更多人!

python学习笔记
https://minikou.cloud/posts/python/
作者
minikou
发布于
2026-05-07
许可协议
Unlicensed

部分信息可能已经过时

目录