两个异步例子。
2026/1/29大约 2 分钟
两个异步例子。
若一个函数是async的,它await时,只会让出cpu,但没法说让它所在的项目中其它任务进行,对吗,也就是没法进行项目内并发(使用asyncio)。
若使用asyncio.gather()的话,则可以在项目内某个环节执行并发。
这种现象是因为一个项目除非使用threading ,否则它就是一条线顺序执行,也就是单线程
import json
with open("data.json", "r", encoding="utf-8") as f:
data = json.load(f)
print(type(data)) # 一般是 dict 或 list
print(data)
import json
def read_jsonl(path):
data = []
with open(path, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
data.append(json.loads(line))
return data
# 用法
records = read_jsonl("data.jsonl")
print(len(records))
print(records[0])
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def gpt_chat(prompt, system_prompt=None, history_messages=[], model_name="openai/gpt-4o-mini"):
messages = []
if system_prompt:
messages.append({"role": "system", "content": system_prompt})
messages.extend(history_messages)
messages.append({"role": "user", "content": prompt})
completion = await client.chat.completions.create(
model=model_name,
messages=messages,
extra_body={}
)
return completion.choices[0].message.content
results = await asyncio.gather(
gpt_chat("Q1", sys_prompt),
gpt_chat("Q2", sys_prompt),
gpt_chat("Q3", sys_prompt),
)
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(filename)s - %(levelname)s - %(message)s'
)
csv 模块import csv
rows = [
["id", "name", "score"],
[1, "Alice", 95],
[2, "Bob", 88],
[3, "Charlie", 92],
]
with open("output.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerows(rows)
print("CSV 写入完成")
import json
data = {
"name": "Alice",
"age": 25,
"skills": ["Python", "AI", "RAG"],
"active": True
}
with open("data.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
import networkx as nx
# 无向图
G = nx.Graph()
# 有向图
DG = nx.DiGraph()
# 多重图
MG = nx.MultiGraph()
# 多重有向图
MDG = nx.MultiDiGraph()
map vs zipa = [1, 2, 3]
b = [4, 5, 6]
# map + lambda
result_map = list(map(lambda x, y: x + y, a, b))
print("map结果:", result_map) # [5, 7, 9]
# zip + 列表推导式
result_zip = [x + y for x, y in zip(a, b)]
print("zip结果:", result_zip) # [5, 7, 9]
# 直接查看zip的结果
print("zip对象:", list(zip(a, b))) # [(1, 4), (2, 5), (3, 6)]