档记录了 Python 多进程与多线程的核心知识点。内容包括使用 Process 类和 Pool 类创建进程，通过 Queue、Pipe、Manager 及 Array 实现进程间通信，利用 Lock 和 Event 进行进程同步。线程部分涵盖 threading 类、派生类、线程池（threadpool 和 concurrent.futures）以及线程通信与同步机制。最后通过 Excel 导入数据库和彩票模拟案例展示了并发编程的实际应用场景。
Multiprocessing 模块中创建进程的类主要有两个：Process 类（进程类）和 Pool 类（进程池类）
start 是创建并启动子进程的函数，join 是等待子进程结束的函数。如果不写 join，可能导致主进程在子进程标识输出前结束。
sp1.join() 检查 sp1 是否还在运行，如果还在运行，主进程进入等待状态。操作系统需要把 CPU 让给 sp1，等它结束后主进程再结束。
from multiprocessing import Process
os
():
( % (spname, os.getpid()))
__name__ == :
()
( % os.getpid())
sp1 = Process(target=spfunc, args=(,))
sp2 = Process(target=spfunc, args=(,))
sp1.start()
sp2.start()
sp1.join()
sp2.join()
()
使用 Process 类的派生类创建进程：
import os, time
from multiprocessing import Process
class SubProcess(Process):
def __init__(self):
Process.__init__(self)
def run(self):
print("子进程名称 = {0}，ID = {1}".format(self.name, self.pid))
time.sleep(0.1)
if __name__ == '__main__':
print("主进程开始：")
for i in range(3):
sp = SubProcess()
sp.start()
sp.join()
print("主进程结束。")
如果要启动大量子进程，可以用 Pool 类的方式批量创建子进程。
import multiprocessing, time, os
def spfunc(i):
print("子进程 ID = {}，结果 = {}".format(os.getpid(), i**2))
time.sleep(1)
if __name__ == '__main__':
pool = multiprocessing.Pool(processes=3)
for i in range(1, 4):
pool.apply(spfunc, (i,))
pool.close()
pool.join()
from multiprocessing import Process, Queue
import os, time, random
def writeQueue(q):
for i in range(5):
n = random.randint(1, 100)
print("放进队列的随机数为：%d" % n)
q.put(n)
time.sleep(1)
def readQueue(q):
for i in range(5):
d = q.get(True)
print("从队列中读取的数据为：%d" % d)
if __name__ == '__main__':
q = Queue()
spw = Process(target=writeQueue, args=(q,))
spr = Process(target=readQueue, args=(q,))
spw.start()
spr.start()
spw.join()
spr.join()
from multiprocessing import Pipe, Process
def spfunc(cPipe):
while True:
try:
print("接受的信息：", cPipe.recv())
except EOFError:
break
if __name__ == '__main__':
parentPipe, childPipe = Pipe(True)
sp = Process(target=spfunc, args=(childPipe,))
sp.start()
parentPipe.send("量子通信")
parentPipe.send("引力波")
parentPipe.close()
sp.join()
from multiprocessing import Process, Manager
def spfunc(dict_obj, list_obj):
dict_obj['Alice'] = 21
dict_obj['Beth'] = 22
list_obj.append("physics")
list_obj.append("chemistry")
if __name__ == '__main__':
manager = Manager()
list_obj = manager.list()
dict_obj = manager.dict()
sp = Process(target=spfunc, args=(dict_obj, list_obj))
sp.start()
sp.join()
print("list:", list_obj)
print("dict:", dict_obj)
关键区别在于共享内存。直接定义列表无法实现共享，需使用 multiprocessing.Array。
import multiprocessing
def spfunc(num):
for i in range(3):
num[i] **= 3
if __name__ == '__main__':
num = multiprocessing.Array("i", [1, 3, 5])
print("初始数组：", num[:])
sp = multiprocessing.Process(target=spfunc, args=(num,))
sp.start()
sp.join()
print("修改后的数组：", num[:])
上锁：一个进程执行完再执行另一个。
sp1:number=1 sp1:number=4 sp1:number=9 sp2:number=1 sp2:number=8 sp2:number=27
把锁注释掉：两个进程交替打印，互相干扰。
sp1:number=1 sp2:number=1 sp1:number=4 sp2:number=8 sp1:number=9 sp2:number=27
from multiprocessing import Process, Lock
import time, os
def spfunc(spname, v, lock):
lock.acquire()
for i in range(1, 4):
n = i**v
time.sleep(0.5)
print("{0}:number={1}".format(spname, n))
lock.release()
if __name__ == "__main__":
lock = Lock()
sp1 = Process(target=spfunc, args=("sp1", 2, lock))
sp2 = Process(target=spfunc, args=("sp2", 3, lock))
sp1.start()
sp2.start()
sp1.join()
sp2.join()
set: V 操作，wait: P 操作，clear：重置操作。
准备买商品... 下订单、付款... 接收订单和贷款... 发货... 收到商品... 评价商品... 欢迎再次订购
import multiprocessing, time
def sjfunc(event):
print("接收订单和贷款...")
print("发货...")
event.set()
event.clear()
event.wait()
print("欢迎再次订购")
def gkfunc(event):
print("准备买商品...")
print("下订单、付款...")
event.wait()
print("收到商品...")
print("评价商品...")
event.set()
event.clear()
if __name__ == '__main__':
event = multiprocessing.Event()
sjp = multiprocessing.Process(target=sjfunc, args=(event,))
gkp = multiprocessing.Process(target=gkfunc, args=(event,))
gkp.start()
time.sleep(0.5)
sjp.start()
from threading import Thread
import time
def thfunc(i):
print("线程%d启动" % i)
time.sleep(1)
if __name__ == '__main__':
print("主线程开始...")
for i in range(3):
th = Thread(target=thfunc, args=(i,))
th.start()
th.join()
print("主线程结束。")
import time, threading
class mythread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self):
for i in range(3):
print("这里是线程%s中的%d" % (self.getName(), i))
time.sleep(1)
if __name__ == '__main__':
th1 = mythread()
th2 = mythread()
th1.start()
th2.start()
th1.join()
th2.join()
threadpool 模块：
线程池有 3 个线程，刚好有 3 个任务
线程 1：执行 addfunc(1, 2) → 输出 "1+2=3" 线程 2：执行 addfunc(3, 4) → 输出 "3+4=7" 线程 3：执行 addfunc(5, 6) → 输出 "5+6=11"
由于有 3 个线程，3 个任务同时执行
输出顺序可能不同，取决于哪个线程先完成
import threadpool
def addfunc(op1, op2):
print("%d+%d=%d" % (op1, op2, op1 + op2))
if __name__ == '__main__':
oplist1 = [1, 2]
oplist2 = [3, 4]
oplist3 = [5, 6]
oplist = [(oplist1, None), (oplist2, None), (oplist3, None)]
pool = threadpool.ThreadPool(3)
requests = threadpool.makeRequests(addfunc, oplist)
for req in requests:
pool.putRequest(req)
pool.wait()
concurrent.futures 模块创建线程池
map 和 submit 都是输出函数。
from concurrent.futures import ThreadPoolExecutor
def executetask(task):
print("执行:%s" % (task))
if __name__ == '__main__':
tasklist = []
for i in range(5):
tasklist.append("任务" + str(i))
with ThreadPoolExecutor(5) as ect:
future = ect.map(executetask, tasklist)
import threading, queue, time
class mythread(threading.Thread):
def __init__(self, q, i):
super(mythread, self).__init__()
self.q = q
self.i = i
def run(self):
self.q.put("这是第%d个线程..." % (self.i))
if __name__ == '__main__':
q = queue.Queue()
for i in range(3):
mythread(q, i).start()
while not q.empty():
print(q.get())
加了锁都是线程 A，不加锁如下。
当前线程是 A,num=1 当前线程是 B,num=2 当前线程是 A,num=3
import threading, time
def countfunc(tname, lock):
global num
lock.acquire()
while True:
if num <= 3:
print("当前线程是%s,num=%s" % (tname, num))
num += 1
time.sleep(1)
else:
break
lock.release()
if __name__ == '__main__':
num = 1
lock = threading.Lock()
t1 = threading.Thread(target=countfunc, args=('A', lock))
t2 = threading.Thread(target=countfunc, args=('B', lock))
t1.start()
t2.start()
t1.join()
t2.join()
import threading, time
def thfunc1(thname, cond):
with cond:
for i in range(0, 6, 2):
print("线程%s:%d." % (thname, i))
cond.wait()
cond.notify()
def thfunc2(thname, cond):
with cond:
for i in range(1, 6, 2):
print("线程%s:%d." % (thname, i))
cond.notify()
cond.wait()
if __name__ == '__main__':
cond = threading.Condition()
th1 = threading.Thread(target=thfunc1, args=("th1", cond))
th2 = threading.Thread(target=thfunc2, args=("th2", cond))
th1.start()
th2.start()
th1.join()
th2.join()
event.isSet() or event.set()是判断是否为 false，如果是 false 那么改成 true，也就是有事件发生。 event.wait() # 等待第一个事件（加班通知），等待第二个事件（吃夜宵通知）。
import time, threading
def bossfunc(thname, event):
print("%s：今晚加班到 11 点" % thname)
event.set()
time.sleep(2)
print("%s：大家一起吃夜宵" % thname)
event.set()
def employeefunc(thname, i, event):
event.wait()
print("%s %d：好痛苦..." % (thname, i))
time.sleep(1)
event.clear()
event.wait()
print("%s %d；太好了" % (thname, i))
if __name__ == '__main__':
event = threading.Event()
thboss = threading.Thread(target=bossfunc, args=("thBoss", event)).start()
for i in range(4):
themployee = threading.Thread(target=employeefunc, args=("thEmployee", i, event)).start()
使用多进程将多个 excel 文件的书籍信息导入 SQLite3 数据库
import multiprocessing, sqlite3, os, openpyxl
def create_connect_database():
try:
conn = sqlite3.connect(os.getcwd() + '\\resources\\bookDB.db')
cur = conn.cursor()
cur.execute(
"""create table if not exists table_book(ID text PRIMARY KEY,Name text not null,Autore text,Price float);"""
)
conn.commit()
except Exception as e:
print("创建/链接数据库或者创建数据表失败:", e)
def eachxlsx(xlsxfn):
wb = openpyxl.load_workbook(xlsxfn)
ws = wb.worksheets[0]
for index, row in enumerate(ws.rows):
if index == 0:
continue
yield tuple(map(lambda x: x.value, row))
def spfunc(filename):
with sqlite3.connect(os.getcwd() + '\\resources\\bookDB.db') as conn:
try:
cur = conn.cursor()
sql = 'INSERT INTO table_book VALUES(?,?,?,?)'
cur.executemany(sql, eachxlsx(filename))
conn.commit()
except Exception as e:
print("导入数据失败.", e)
if __name__ == '__main__':
print("Excel 文件导入数据库开始...")
create_connect_database()
pool = multiprocessing.Pool()
excel_path = os.getcwd() + "\\excel_files_dir"
excel_File_list = os.listdir(excel_path)
for filename in excel_File_list:
excelFile = excel_path + "\\" + filename
pool.apply(spfunc, (excelFile,))
pool.close()
pool.join()
print("Excel 文件数据导入数据库结束！")
zjxx = self.cpfx_q.get() #取出中奖信息，并删除 cphmlist = self.cpfx_q.get() #取出可购买的彩票号码列表，并删除
这种"取出即删除"的设计是为了保证数据不会被多个线程重复处理。
import threading, queue, random
class mythread(threading.Thread):
def __init__(self, cpfx_q, i, zjhm, lock):
super(mythread, self).__init__()
self.cpfx_q = cpfx_q
self.i = i
self.zjhm = zjhm
self.lock = lock
def run(self):
self.lock.acquire()
zjxx = self.cpfx_q.get()
cphmlist = self.cpfx_q.get()
cphm = random.choice(cphmlist)
cphm_num = cphmlist.index(cphm)
if cphm == self.zjhm:
zjxx = "抽奖人" + str(self.i) + "号购买的彩票" + str(cphm) + "中奖了！恭喜您！"
cphmlist.pop(cphm_num)
self.cpfx_q.put(zjxx)
self.cpfx_q.put(cphmlist)
self.lock.release()
if __name__ == '__main__':
print("————————————————彩票发行开始————————————————")
cpfx_q = queue.Queue()
cphmlist = list(range(1000, 10000))
random.shuffle(cphmlist)
zjxx = "无人中奖！"
cpfx_q.put(zjxx)
cpfx_q.put(cphmlist)
zjhm = random.randrange(1000, 10000)
print("本次发行彩票中奖号码：", zjhm)
lock = threading.Lock()
for i in range(5000):
mythread(cpfx_q, i, zjhm, lock).start()
print("本次彩票发行结果：", cpfx_q.get())
print("————————————————彩票发行结束————————————————")
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online