1. os 和 sys(系統編程基礎)
這兩個模塊是進行系統層面操作(如文件管理、路徑處理、環境變量訪問等)必不可少的工具。
os 模塊
os
主要是用于與操作系統交互的,比如:
-
文件和目錄操作
-
獲取系統信息
-
運行系統命令
-
管理環境變量
可以理解為:讓你的 Python 代碼能夠操作"文件系統"、"目錄結構"、"系統指令"等資源。
路徑操作:
import osprint(os.getcwd()) # 獲取當前工作目錄
os.chdir('./tmp') # 改變當前工作目錄
print(os.listdir('.')) # 列出當前目錄下的所有文件和文件夾 當前工作目錄已經是 ./tmp
文件/文件夾操作:
os.mkdir('testdir') # 創建文件夾
os.rmdir('testdir') # 刪除空文件夾
os.remove('file.txt') # 刪除文件
os.rename('old.txt', 'new.txt') # 重命名文件或文件夾
路徑拼接與判斷:
path = os.path.join('folder', 'file.txt') # 拼接路徑(自動處理/或\問題)
print(os.path.exists(path)) # 檢查路徑是否存在
print(os.path.isfile(path)) # 是否是文件
print(os.path.isdir(path)) # 是否是目錄
執行系統命令:
# 獲取環境變量
home = os.environ.get('HOME')
print(home)os.system('ls -l') # 運行一個系統命令(同步阻塞)
(注:更推薦用 subprocess
,因為更安全、靈活)
sys 模塊
sys
主要是用于跟 Python 解釋器交互,比如:
-
讀取命令行參數
-
退出程序
-
獲取 Python 環境信息
-
修改模塊搜索路徑
可以理解為:讓你的程序能控制/了解運行時自身的狀態。
命令行參數:
import sysprint(sys.argv) # 獲取命令行參數,列表,第一個是程序本身# 例如命令行運行:
# python test.py arg1 arg2
# sys.argv 結果就是:
# ['test.py', 'arg1', 'arg2']
退出程序:
sys.exit(0) # 正常退出
sys.exit(1) # 異常退出
Python 版本信息:
print(sys.version) # 打印Python版本
print(sys.platform) # 當前系統平臺:'win32', 'linux', 'darwin'(Mac)
修改模塊搜索路徑:
import sys# 加了之后,Python在**import 模塊**時,除了默認的系統路徑,還會去你指定的新路徑里找模塊。
sys.path.append('/path/to/your/module') # 動態添加模塊搜索路徑
(比如項目中的一些自定義模塊,需要臨時導入)
實際綜合例子
寫一個 Python 腳本,它可以根據命令行傳參,創建一個目錄并進入:
import os
import sysdef create_and_enter_dir(dirname):if not os.path.exists(dirname):os.mkdir(dirname)print(f"Directory '{dirname}' created.")else:print(f"Directory '{dirname}' already exists.")os.chdir(dirname)print(f"Now in directory: {os.getcwd()}")if __name__ == "__main__":if len(sys.argv) != 2:print("Usage: python script.py <dirname>")sys.exit(1)create_and_enter_dir(sys.argv[1])
這樣可以:
python script.py myfolder
自動創建并進入 myfolder
。
2. datetime(時間處理)
用于處理日期和時間的模塊,常用于日志、時間戳生成、時間間隔計算等。
獲取當前時間:
import datetimenow = datetime.datetime.now()
print(now) # 比如輸出:2025-04-29 15:30:42.123456
?如果只要日期
today = datetime.date.today()
print(today) # 比如:2025-04-29
創建指定時間:
d = datetime.datetime(2025, 5, 1, 12, 30, 45) # 年, 月, 日, 時, 分, 秒
print(d) # 2025-05-01 12:30:45
如果只要日期對象
d = datetime.date(2025, 5, 1)
print(d) # 2025-05-01
時間加減(使用 timedelta
),比如1天后、2小時前、10分鐘后:
timedelta
支持的參數有:days
, seconds
, microseconds
, milliseconds
, minutes
, hours
, weeks
from datetime import datetime, timedeltanow = datetime.now()print(now + timedelta(days=1)) # 一天后
print(now - timedelta(hours=2)) # 兩小時前
print(now + timedelta(minutes=10)) # 十分鐘后
時間格式化輸出(變成字符串)
比如格式化成 "2025-04-29 15:30:42"
這種:
now = datetime.datetime.now()print(now.strftime("%Y-%m-%d %H:%M:%S"))
# 輸出:2025-04-29 15:30:42
字符串轉時間對象(解析時間)
如果拿到的是字符串,比如 "2025-04-29 15:30:42"
,想轉成 datetime
對象:
(注意 strptime
和 strftime
名字很像,p 是 parse(解析))
s = "2025-04-29 15:30:42"
dt = datetime.datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
print(dt)
# 輸出:2025-04-29 15:30:42
時間比較
可以直接比較兩個 datetime
對象:
t1 = datetime.datetime(2025, 5, 1, 12, 0, 0)
t2 = datetime.datetime(2025, 5, 2, 12, 0, 0)print(t1 < t2) # True
print(t2 - t1) # 1 day, 0:00:00 (是一個 timedelta)
循環生成時間段(進階常用)
比如想生成從 2025-04-29 00:00 開始,每隔1小時一個點:
start = datetime.datetime(2025, 4, 29, 0, 0, 0)
end = datetime.datetime(2025, 4, 29, 5, 0, 0)current = start
while current <= end:print(current.strftime("%Y-%m-%d %H:%M:%S"))current += datetime.timedelta(hours=1)
要處理時區,就要用到 tzinfo
,不過一般推薦配合第三方庫 pytz
或 zoneinfo
來做。
from datetime import datetime, timezone, timedelta# 設置時區:UTC+8
dt = datetime(2025, 4, 29, 12, 0, 0, tzinfo=timezone(timedelta(hours=8)))
print(dt)
3. re(正則表達式)
正則表達式(Regular Expression)是一種字符串匹配規則的描述語言。
它能非常靈活地搜索、提取、替換字符串中的特定內容,比如:
-
驗證手機號
-
提取網頁上的郵箱
-
替換文本中的敏感詞
-
分割復雜字符串
-
檢測密碼復雜度
Python 處理正則,靠的就是內置的 re
模塊。
re
模塊常用函數
導入:
import re
常用函數有:
函數 | 作用 |
---|---|
re.match(pattern, string) | 從字符串開頭匹配 |
re.search(pattern, string) | 搜索整個字符串,找到第一個匹配 |
re.findall(pattern, string) | 找到所有匹配,返回列表 |
re.finditer(pattern, string) | 找到所有匹配,返回迭代器(內存更省) |
re.sub(pattern, repl, string) | 替換匹配到的內容 |
re.split(pattern, string) | 按正則分割字符串 |
re.compile(pattern) | 編譯成正則對象,提高效率 |
正則表達式語法基礎
字面意義:
普通字符,自己代表自己,比如 abc
就匹配 abc
。
元字符(特殊意義的字符):
字符 | 含義 |
---|---|
. | 任意一個字符(除了換行) |
^ | 開頭匹配 |
$ | 結尾匹配 |
* | 重復0次或更多次 |
+ | 重復1次或更多次 |
? | 重復0次或1次 |
{n} | 恰好n次 |
{n,} | 至少n次 |
{n,m} | n到m次 |
[...] | 匹配字符集中的任意一個字符 |
` | ` |
() | 分組(提取、捕獲) |
轉義字符(\
)
如果你想匹配特殊字符,比如? ?.?
,必須用? ?\.?
。
比如:
re.match(r"a\.b", "a.b") # 匹配成功
注意:正則表達式字符串一般加 r""
,表示原生字符串,不轉義。
常用字符集
表達式 | 含義 |
---|---|
\d | 數字,等價于[0-9] |
\D | 非數字 |
\w | 字母數字下劃線,等價于[a-zA-Z0-9_] |
\W | 非字母數字下劃線 |
\s | 空白字符(空格、tab、換行) |
\S | 非空白字符 |
貪婪 vs 非貪婪
-
正則默認是貪婪匹配(能多就多)
-
在量詞后加
?
,變成非貪婪匹配(能少就少)
例子:
import retext = "<p>Title</p><p>Content</p>"print(re.findall(r"<p>.*</p>", text)) # 貪婪,匹配到整個 "<p>Title</p><p>Content</p>"
print(re.findall(r"<p>.*?</p>", text)) # 非貪婪,分別匹配 "<p>Title</p>" 和 "<p>Content</p>"
典型應用案例
匹配手機號
import rephone = "我的號碼是13812345678,你的呢?"pattern = r"1[3-9]\d{9}"result = re.search(pattern, phone)
print(result.group()) # 輸出:13812345678
提取網頁中的所有郵箱地址
text = "聯系:user1@gmail.com, admin@company.cn, feedback@yahoo.com"emails = re.findall(r"[a-zA-Z0-9._%-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", text)
print(emails)
# 輸出:['user1@gmail.com', 'admin@company.cn', 'feedback@yahoo.com']
替換敏感詞
text = "這是一段包含badword1和badword2的文字"pattern = r"badword1|badword2"
clean_text = re.sub(pattern, "***", text)
print(clean_text)
# 輸出:這是一段包含***和***的文字
分割復雜字符串
比如有一段文字,逗號、分號、空格都可能作為分隔:
text = "apple,banana;orange pear"words = re.split(r"[,;\s]+", text)
print(words)
# 輸出:['apple', 'banana', 'orange', 'pear']
分組提取
-
在正則表達式里,用小括號
()
把一部分內容包起來。 -
這樣就可以單獨提取出括號里的子匹配。
-
每個括號對應一個組(group),編號從 1 開始。
.group(0)
是整個匹配到的字符串;.group(1)
, .group(2)
分別是第1個、第2個括號匹配到的內容。
import retext = '姓名:張三,電話:13812345678'pattern = r'姓名:(.*),電話:(1[3-9]\d{9})'result = re.search(pattern, text)print(result.group(0)) # 整個匹配:姓名:張三,電話:13812345678
print(result.group(1)) # 第1組:(.*) -> 張三
print(result.group(2)) # 第2組:(1[3-9]\d{9}) -> 13812345678
compile優化
如果一個正則要重復用多次,可以預編譯成對象:
好處:效率更高,尤其是循環里用的時候。
pattern = re.compile(r"\d+")
print(pattern.findall("123 abc 456"))
總結:re
模塊讓你可以用一套靈活的規則(正則表達式),在字符串中快速定位、提取、替換內容,非常強大。
4. json(數據序列化)
JSON(JavaScript Object Notation)是一種輕量級的數據交換格式,特點是:
-
結構簡單(鍵值對、數組)
-
可讀性好(人和機器都能輕松理解)
-
跨語言通用(不只是 JS,Python/Java/Go/后端接口全用)
Python對象和JSON格式互轉,非常適合與前后端接口對接、配置文件讀寫。
樣例 JSON:
{"name": "小明","age": 24,"skills": ["Python", "Java", "Hacking"],"has_job": false
}
常用于:
-
后端接口返回(API)
-
配置文件(如
.json
結尾) -
本地保存數據(緩存)
-
網絡通信
核心函數用法
Python 提供了內置的 json
模塊,主要做兩件事:
操作 | 函數 | 作用 |
---|---|---|
Python 對象 → JSON 字符串 | json.dumps() | 序列化(dump string) |
JSON 字符串 → Python 對象 | json.loads() | 反序列化(load string) |
Python 對象 → 寫入 JSON 文件 | json.dump() | 序列化并寫入文件 |
JSON 文件 → Python 對象 | json.load() | 反序列化并讀取文件 |
json.dumps()
把 Python 對象轉換成 JSON 字符串。
import jsondata = {"name": "小明", "age": 24, "skills": ["Python", "Hacking"]}json_str = json.dumps(data)
print(json_str)
# 輸出:{"name": "\u5c0f\u660e", "age": 24, "skills": ["Python", "Hacking"]}
默認中文會轉成 Unicode 編碼!
想讓中文正常顯示,加參數:
json_str = json.dumps(data, ensure_ascii=False)
print(json_str)
# 輸出:{"name": "小明", "age": 24, "skills": ["Python", "Hacking"]}
美化輸出(縮進):
json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)
輸出:
{"name": "小明","age": 24,"skills": ["Python","Hacking"]
}
json.loads()
把 JSON 字符串轉換成 Python 對象。
json_str = '{"name": "小明", "age": 24, "skills": ["Python", "Hacking"]}'data = json.loads(json_str)
print(data)
# 輸出:{'name': '小明', 'age': 24, 'skills': ['Python', 'Hacking']}
json.dump()
把 Python 對象寫入到文件中(保存成 .json
文件)。
data = {"name": "小明", "age": 24}with open('data.json', 'w', encoding='utf-8') as f:json.dump(data, f, ensure_ascii=False, indent=2)
寫文件時注意 encoding='utf-8'
,否則中文可能亂碼。
json.load()
從文件中讀取 JSON 數據,變成 Python 對象。
with open('data.json', 'r', encoding='utf-8') as f:data = json.load(f)
print(data)
Python類型和JSON類型的對應關系
Python類型 | JSON類型 |
---|---|
dict | 對象(Object) |
list 、tuple | 數組(Array) |
str | 字符串(String) |
int 、float | 數字(Number) |
True 、False | 布爾值(true/false) |
None | 空(null) |
注意:Python 的 None
會變成 JSON 的 null
。
常見應用場景示例
處理 API 接口返回的 JSON
import requests
import jsonresp = requests.get('https://api.example.com/data')
data = json.loads(resp.text)
print(data)
保存爬蟲數據
results = [{"title": "新聞1"}, {"title": "新聞2"}]with open('results.json', 'w', encoding='utf-8') as f:json.dump(results, f, ensure_ascii=False, indent=2)
配置文件讀取
with open('config.json', 'r', encoding='utf-8') as f:config = json.load(f)print(config['api_key'])
總結:json
模塊就是把Python數據和JSON字符串互相轉換的橋梁,非常適合用來做接口交互、數據保存、配置管理。
5. collections(高級數據結構)
collections
是 Python 標準庫里的一個模塊。
專門提供比內置數據結構(如 list、dict)更強大、更高效的“容器類”工具。
比如:
-
更智能的字典(
defaultdict
、OrderedDict
) -
可統計的計數器(
Counter
) -
支持隊列操作的雙端隊列(
deque
) -
可以給元組加字段名(
namedtuple
) -
專業的堆棧/鏈表(
ChainMap
)
常用的有這幾個:
名稱 | 簡單理解 | 作用 |
---|---|---|
namedtuple | 有名字的元組 | 類似輕量版對象(只有屬性沒有方法) |
deque | 雙端隊列 | 高效插入/刪除元素(兩邊都快) |
Counter | 計數器 | 統計元素出現的次數 |
OrderedDict | 有序字典(老版本用) | 按插入順序記住鍵值對(Python 3.7+內置dict已有序) |
defaultdict | 帶默認值的字典 | 取不存在的key不會報錯,返回默認值 |
ChainMap | 多字典組合 | 多個字典邏輯上合并查詢 |
namedtuple
(有名字的元組)
像 tuple
,但可以用名字訪問元素,更直觀。
用法場景:臨時定義小對象,不想寫復雜 class
from collections import namedtuple# 定義一個點
Point = namedtuple('Point', ['x', 'y'])p = Point(1, 2)
print(p.x, p.y) # 1 2
print(p) # Point(x=1, y=2)
特點:
-
占用內存小
-
代碼簡潔
-
支持索引訪問,也支持屬性名訪問
deque
(雙端隊列)
比 list 更高效,兩邊都能快速添加/刪除元素。
用法場景:需要頻繁在頭部/尾部插入或刪除,比如 BFS、滑動窗口。
from collections import dequeq = deque([1, 2, 3])q.append(4) # 末尾加
q.appendleft(0) # 開頭加
print(q) # deque([0, 1, 2, 3, 4])q.pop() # 刪除末尾
q.popleft() # 刪除開頭
print(q) # deque([1, 2, 3])
特性:
-
頭尾操作都是 O(1) 復雜度
-
比 list.insert(0, x) 高效得多
Counter
(計數器)
用來統計元素出現的次數。
用法場景:數據統計、文本詞頻分析。
from collections import Counterc = Counter('abracadabra')
print(c)
# Counter({'a': 5, 'b': 2, 'r': 2, 'c': 1, 'd': 1})print(c.most_common(2)) # 出現最多的2個元素
# [('a', 5), ('b', 2)]
特性:
-
可以直接加減集合
-
支持最常見元素提取
OrderedDict
(有序字典)
普通 dict
在 Python 3.7+ 默認已經有序,但在老版本(3.6以前)中,OrderedDict
很重要。
用法場景:需要維護元素插入順序。
from collections import OrderedDictod = OrderedDict()
od['a'] = 1
od['b'] = 2
od['c'] = 3print(od) # OrderedDict([('a', 1), ('b', 2), ('c', 3)])
特性:
-
遍歷時,順序和插入順序一致。
defaultdict
(帶默認值的字典)
普通 dict
訪問不存在的 key 會報錯。defaultdict
自動給一個默認值。
用法場景:構建復雜字典(比如分組統計、字典嵌套)
from collections import defaultdictd = defaultdict(int) # 默認值是 int(), 即0
d['a'] += 1
print(d['a']) # 1
print(d['b']) # 0 (不會報錯)# 嵌套使用
tree = defaultdict(lambda: defaultdict(list))
tree['python']['versions'].append(3.9)
print(tree)
特性:
-
自動初始化新key
-
很適合做分組/聚合操作
ChainMap
(字典鏈)
多個字典組合在一起,查詢時像一個整體。
用法場景:多層配置覆蓋,比如環境變量優先級。
from collections import ChainMapdefaults = {'theme': 'Default', 'language': 'English'}
overrides = {'theme': 'Dark'}settings = ChainMap(overrides, defaults)
print(settings['theme']) # Dark(優先查overrides)
print(settings['language']) # English(fallback到defaults)
特性:
-
查詢時,優先從前面的字典查
-
不是真的合并,只是邏輯組合,非常高效
常見使用場景總結
場景 | 用什么 |
---|---|
快速建小對象(無方法的) | namedtuple |
隊列、棧、滑動窗口 | deque |
統計元素出現次數 | Counter |
需要順序的字典 | OrderedDict |
字典默認值處理 | defaultdict |
多層配置管理 | ChainMap |
6. logging(日志模塊)
logging
是 Python 的標準日志模塊,主要功能是:
把程序運行時的關鍵信息輸出到:
-
控制臺(屏幕)
-
文件(保存起來)
-
網絡、郵件、數據庫等(更高級)
讓你:
-
不再亂用
print
-
統一管理日志
-
區分不同重要程度的日志(比如正常信息 vs 錯誤信息)
-
定制輸出格式、保存位置、保存數量
logging
是專業的 "程序黑匣子",記錄程序的一舉一動,出問題能迅速回溯!
logging
內部有四大核心對象:
名稱 | 作用 |
---|---|
Logger | 產生日志的對象(你用的入口) |
Handler | 決定日志去哪兒(控制臺?文件?郵箱?) |
Formatter | 決定日志長什么樣(格式) |
Filter | 決定日志過不過濾(可選,不常用) |
Logger(負責發日志)↓
Handler(負責把日志發到地方)↓
Formatter(控制日志格式)
每個 Logger 可以有多個 Handler,
每個 Handler 都可以指定自己的 Formatter。
比如同時打印到屏幕 + 保存到文件,但格式可以不一樣!
logging常用功能
你常用的基本是這幾種功能:
-
logging.debug()
—— 調試信息 -
logging.info()
—— 普通運行信息 -
logging.warning()
—— 警告(小問題) -
logging.error()
—— 錯誤(程序還能跑) -
logging.critical()
—— 致命錯誤(要崩)
還能做:
-
設置日志等級(控制打印多少)
-
保存到日志文件
-
定時輪換日志(比如每天生成一個新日志)
-
按大小輪換日志(比如100MB一個文件)
-
設置日志格式(時間、模塊、行號等信息)
logging的五個等級
等級 | 函數 | 用途 | 默認級別 |
---|---|---|---|
DEBUG | logging.debug | 最詳細的調試信息 | |
INFO | logging.info | 普通的信息,比如程序開始了、結束了 | |
WARNING | logging.warning | 警告,程序還能繼續運行 | ? |
ERROR | logging.error | 錯誤,影響功能 | |
CRITICAL | logging.critical | 嚴重錯誤,程序可能崩潰 |
注意!
默認的日志級別是 WARNING
,所以低于 WARNING(比如 DEBUG、INFO)不會顯示,除非你自己改級別!
常見配置方式(兩種)
基本配置
import logginglogging.basicConfig(level=logging.INFO,format="%(asctime)s [%(levelname)s] %(message)s",
)logging.info("程序啟動了")
logging.warning("小心,有點不對勁")
logging.error("程序出錯了!")
輸出示例:
2025-04-29 20:35:00 [INFO] 程序啟動了
2025-04-29 20:35:01 [WARNING] 小心,有點不對勁
2025-04-29 20:35:02 [ERROR] 程序出錯了!
basicConfig
可以快速設置:
-
level
:日志等級 -
format
:日志格式 -
filename
:寫到哪個文件 -
filemode
:寫入模式('w'覆蓋,'a'追加)
更復雜的手動配置
import logginglogger = logging.getLogger('my_logger') # 創建一個Logger
logger.setLevel(logging.DEBUG) # 設置日志級別# 創建Handler
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler('my_log.log', encoding='utf-8')# 創建Formatter
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')# 綁定Formatter到Handler
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)# 綁定Handler到Logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)# 開始使用
logger.info('這是一條info日志')
logger.error('這是一條error日志')
這樣可以:
-
控制臺有一份
-
文件保存一份
-
格式一致
實戰例子
目標 | 寫法示例 |
---|---|
設置日志保存到文件 | filename='xxx.log' |
保存中文日志 | encoding='utf-8' |
每天生成一個新日志 | 用 TimedRotatingFileHandler |
文件太大自動切分 | 用 RotatingFileHandler |
改變日志輸出樣式 | 自定義format |
日志按大小切割:
from logging.handlers import RotatingFileHandler
import logginghandler = RotatingFileHandler('big.log', maxBytes=1024*1024*5, backupCount=3)
logging.basicConfig(handlers=[handler], level=logging.DEBUG, format='%(asctime)s %(levelname)s %(message)s')logging.info("這是一個測試日志")
-
單個日志最大5MB
-
最多保留3個歷史文件(big.log.1, big.log.2, big.log.3)
總結:logging = 專業版print,程序運行過程的黑匣子,調試、追錯、歸檔必備神器!
7. subprocess(執行系統命令)
更安全、更強大、更靈活的執行外部命令方法。
簡單來說:subprocess
= 幫你在Python里開一個子進程(子命令行),執行命令,拿到輸出。
流程:
Python程序↓
subprocess模塊↓
操作系統(開子進程,執行系統命令)↓
拿到返回值、輸出結果
subprocess.run()
基礎用法
基本格式:
import subprocesssubprocess.run(["命令", "參數1", "參數2", ...])
例子:
import subprocess# 在Windows上執行 dir
subprocess.run(["dir"], shell=True)# 在Linux上執行 ls -l
subprocess.run(["ls", "-l"])
注意:
-
shell=True
:讓命令像直接在終端里那樣執行。 -
默認
shell=False
,更安全。
捕獲輸出(stdout / stderr)
想拿到命令輸出?用 capture_output=True
:
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)
print(result.stdout) # 正常輸出
print(result.stderr) # 錯誤輸出
參數解釋:
-
capture_output=True
:自動收集輸出。 -
text=True
:讓輸出是字符串(默認是字節bytes)。
傳入參數(列表 vs 字符串)
寫法 | 說明 |
---|---|
列表形式 | 推薦!安全!每個參數獨立 |
字符串形式 | 需要加 shell=True ,容易命令注入(危險) |
列表寫法(推薦):
subprocess.run(["ping", "www.baidu.com"])
字符串寫法(小心):
subprocess.run("ping www.baidu.com", shell=True)
如果要執行復雜管道命令(比如grep、|管道),必須用 shell=True
。
更底層的用法(Popen)
如果想實時讀輸出、或者同時操作輸入輸出,用 Popen
。
簡單例子:
import subprocessp = subprocess.Popen(["ping", "www.baidu.com"], stdout=subprocess.PIPE, text=True)for line in p.stdout:print(line.strip())
這樣可以一行行讀輸出,非常適合:
-
實時監控命令輸出
-
做管道(pipe)連接多個命令
常見錯誤和注意事項
錯誤 | 解釋 |
---|---|
FileNotFoundError | 命令打錯了,找不到 |
PermissionError | 沒權限執行(尤其是Linux) |
UnicodeDecodeError | 輸出是二進制,不是文本,加 text=True |
命令執行失敗 | 查看 result.returncode != 0,說明出錯了 |
好習慣:
result = subprocess.run(["ls", "-l"], capture_output=True, text=True)if result.returncode != 0:print("命令執行失敗!")print(result.stderr)
else:print(result.stdout)
實戰例子(批量執行 / 調用shell腳本)
批量執行一堆命令:
commands = [["echo", "hello"],["ls", "-l"],["pwd"],
]for cmd in commands:subprocess.run(cmd)
調用本地shell腳本:
比如有一個 test.sh
文件:
#!/bin/bash
echo "Hello from script"
Python執行:
subprocess.run(["bash", "test.sh"])
或者:
subprocess.run("./test.sh", shell=True)
注意腳本要有執行權限(chmod +x test.sh
)。
總結
subprocess.run() - 快速執行,拿返回值
subprocess.Popen()- 高級控制,實時操作,讀寫管道
shell=True - 字符串形式(有風險)
capture_output=True- 捕獲輸出
text=True- 輸出是字符串
8. 并發與并行(concurrent.futures、threading、multiprocessing)
概念 | 解釋 | 打比方 |
---|---|---|
并發(Concurrency) | 同一時間段內處理多個任務(假裝同時做,其實輪著做) | 一個人炒兩個菜,輪流翻鍋 |
并行(Parallelism) | 真正同時處理多個任務(多核心同時進行) | 兩個人同時炒兩個菜 |
并發是邏輯上的同時(比如單核CPU上下文切換)
并行是物理上的同時(比如多核CPU各自干活)
Python中的并發并行手段
模塊 | 適用場景 | 特點 |
---|---|---|
threading | I/O密集型任務(網絡、磁盤讀寫) | 多線程,輕量,但受GIL限制 |
multiprocessing | CPU密集型任務(復雜計算) | 多進程,真正并行,開銷大 |
concurrent.futures | 更高級的線程池/進程池 | 簡單易用,高層封裝 |
GIL(全局解釋器鎖)讓多線程不能真正并行執行Python代碼,但遇到I/O會釋放鎖。
多進程可以真正利用多核CPU,每個進程都有獨立的Python解釋器。
threading(多線程模塊)
主要用于 I/O密集型任務:比如抓取網頁、文件讀寫、網絡請求。
基本用法
import threadingdef worker():print("我是子線程")# 創建線程
t = threading.Thread(target=worker)# 啟動線程
t.start()# 等待線程結束
t.join()
-
start()
:啟動線程(真正開始執行) -
join()
:主線程等待子線程結束
多線程執行示例
for i in range(5):threading.Thread(target=worker).start()
多線程是共享內存的,可以共享數據,但要小心數據競爭!
注意事項
-
GIL導致多線程在Python解釋器層面不是并行的(CPU密集型反而慢)
-
線程適合用來等待I/O,而不是計算密集的工作
multiprocessing(多進程模塊)
主要用于 CPU密集型任務:比如數據處理、圖像處理、大型計算。
4.1 基本用法
import multiprocessingdef worker():print("我是子進程")if __name__ == '__main__':p = multiprocessing.Process(target=worker)p.start()p.join()
-
start()
:啟動進程 -
join()
:等待進程結束
多進程執行示例
if __name__ == '__main__':for i in range(5):multiprocessing.Process(target=worker).start()
每個子進程是完全獨立的,不共享內存!
注意事項
-
多進程開銷大(復制一份整個程序到子進程)
-
需要用
multiprocessing.Queue
等方式進行進程間通信
concurrent.futures(高級封裝庫)
Python3之后出的一個超好用模塊,
封裝了線程池 (ThreadPoolExecutor
) 和 進程池 (ProcessPoolExecutor
)。讓并發變得超級簡單!
基本用法(線程池)
from concurrent.futures import ThreadPoolExecutordef worker(x):return x * xwith ThreadPoolExecutor(max_workers=5) as executor:results = executor.map(worker, [1, 2, 3, 4, 5])for r in results:print(r)
-
max_workers
:同時跑幾個線程 -
map
:批量提交任務 -
自動管理
start
、join
,不用手動操心!
基本用法(進程池)
from concurrent.futures import ProcessPoolExecutordef worker(x):return x * xwith ProcessPoolExecutor(max_workers=5) as executor:results = executor.map(worker, [1, 2, 3, 4, 5])for r in results:print(r)
一行切換線程池/進程池,爽爆!
三者的區別與選擇
特性 | threading | multiprocessing | concurrent.futures |
---|---|---|---|
本質 | 線程 | 進程 | 封裝線程池/進程池 |
適合 | I/O密集 | CPU密集 | 兩者皆可 |
是否并行 | 否(受GIL影響) | 是(獨立進程) | 看用的Executor |
編程復雜度 | 中 | 高(通信麻煩) | 低(傻瓜式) |
內存開銷 | 小 | 大 | 中等 |
簡單記憶:
-
網絡爬蟲、文件下載 →
threading
或ThreadPoolExecutor
-
復雜計算、圖像處理 →
multiprocessing
或ProcessPoolExecutor
-
不想操心細節 →
concurrent.futures
實戰示例(小項目案例)
用線程池并發抓取網頁
import requests
from concurrent.futures import ThreadPoolExecutorurls = ['http://example.com','http://example.org','http://example.net',
]def fetch(url):r = requests.get(url)print(f"{url}: {r.status_code}")with ThreadPoolExecutor(max_workers=3) as executor:executor.map(fetch, urls)
多線程抓網頁,飛快!
用進程池并行計算
from concurrent.futures import ProcessPoolExecutordef calc(n):total = 0for i in range(10**6):total += i * nreturn totalif __name__ == '__main__':with ProcessPoolExecutor() as executor:results = executor.map(calc, range(5))print(list(results))
多核CPU全開,處理大計算量!
總結
如果做的是:
-
抓網頁、爬蟲、IO讀寫、等待數據庫 → 用線程(
threading
/ThreadPoolExecutor
) -
大量數據運算、模型訓練、復雜數學計算 → 用進程(
multiprocessing
/ProcessPoolExecutor
) -
不想自己管理啟動關閉,想一行搞定 → 用
concurrent.futures
注意:
-
threading
快但是受GIL限制,適合I/O密集 -
multiprocessing
真正并行,但開銷更大 -
concurrent.futures
非常適合90%的并發場景
IO密集用線程,CPU密集用進程,怕麻煩就用concurrent.futures。