Python多进程multiprocessing模块核心用法与实战解析
发布时间：2026-10-13| 2人浏览过
本文深入解析Python multiprocessing模块的核心用法，涵盖进程创建、命名与状态跟踪、守护进程配置及终止机制。重点讲解利用Queue实现进程间数据传递，以及Event同步对象在状态共享中的应用，配合代码实例展示多进程编程的最佳实践。
本文通过具体实例详细解析Python中multiprocessing模块的使用方法，旨在为读者提供清晰的参考与指导。
multiprocessing模块简介
该模块的核心功能在于像管理线程一样管理进程，其API设计与threading模块高度相似。相比threading，multiprocessing能更充分地利用多核CPU的性能，提升并行处理效率。
创建进程的基本方法如下：
import multiprocessing
def worker(num):
"""thread worker function"""
print 'Worker:', num
return
if __name__ == '__main__':
jobs = []
for i in range(5):
p = multiprocessing.Process(target=worker, args=(i,))
jobs.append(p)
p.start()
在创建进程时，建议为进程指定名称，以便在运行过程中标识和区分不同进程，便于后续的状态跟踪与调试。
import multiprocessing
import time
def worker():
name = multiprocessing.current_process().name
print name, 'Starting'
time.sleep(2)
print name, 'Exiting'
def my_service():
name = multiprocessing.current_process().name
print name, 'Starting'
time.sleep(3)
print name, 'Exiting'
if __name__ == '__main__':
service = multiprocessing.Process(name='my_service',
target=my_service)
worker_1 = multiprocessing.Process(name='worker 1',
target=worker)
worker_2 = multiprocessing.Process(target=worker) # default name
worker_1.start()
worker_2.start()
service.start()
守护进程（Daemon Process）不会阻止主程序退出，而是独立运行。通过设置multiprocessing.setDaemon(True)（注：实际API中通常通过Process的daemon属性设置）可将其标记为守护进程。若需等待守护进程结束，必须调用join()方法；该方法支持传入浮点数作为超时时间，指定等待时长后自动继续执行。
守护进程的代码示例如下：
import multiprocessing
import time
import sys
def daemon():
name = multiprocessing.current_process().name
print 'Starting:', name
time.sleep(2)
print 'Exiting :', name
def non_daemon():
name = multiprocessing.current_process().name
print 'Starting:', name
print 'Exiting :', name
if __name__ == '__main__':
d = multiprocessing.Process(name='daemon',
target=daemon)
d.daemon = True
n = multiprocessing.Process(name='non-daemon',
target=non_daemon)
n.daemon = False
d.start()
n.start()
d.join(1)
print 'd.is_alive()', d.is_alive()
n.join()
推荐使用“毒丸”（poison pill）机制强制终止进程，即调用terminate()方法。调用terminate()后，务必紧跟join()调用来更新进程状态，确保资源正确释放。
终止进程的示例代码如下：
import multiprocessing import time def slow_worker(): print 'Starting worker' time.sleep(0.1) print 'Finished worker' if __name__ == '__main__': p = multiprocessing.Process(target=slow_worker) print 'BEFORE:', p, p.is_alive() p.start() print 'DURING:', p, p.is_alive() p.terminate() print 'TERMINATED:', p, p.is_alive() p.join() print 'JOINED:', p, p.is_alive()
进程退出状态的判断规则如下： ①. 返回值 == 0 表示进程正常结束，未产生任何错误。 ②. 返回值 > 0 表示进程发生错误，并以该错误码退出。 ③. 返回值 < 0 表示进程被信号终止，其值为 -1 * 发送的信号编号。
检查进程退出状态的示例代码如下：
import multiprocessing
import sys
import time
def exit_error():
sys.exit(1)
def exit_ok():
return
def return_value():
return 1
def raises():
raise RuntimeError('There was an error!')
def terminated():
time.sleep(3)
if __name__ == '__main__':
jobs = []
for f in [exit_error, exit_ok, return_value, raises, terminated]:
print 'Starting process for', f.func_name
j = multiprocessing.Process(target=f, name=f.func_name)
jobs.append(j)
j.start()
jobs[-1].terminate()
for j in jobs:
j.join()
print '%15s.exitcode = %s' % (j.name, j.exitcode)
为了方便调试，建议结合logging模块输出进程运行日志。
日志记录的示例代码如下：
import multiprocessing import logging import sys def worker(): print 'Doing some work' sys.stdout.flush() if __name__ == '__main__': multiprocessing.log_to_stderr() logger = multiprocessing.get_logger() logger.setLevel(logging.INFO) p = multiprocessing.Process(target=worker) p.start() p.join()
可以通过继承multiprocessing.Process类来创建自定义进程，实现更灵活的定制功能。
派生自定义进程类的示例代码如下：
import multiprocessing
class Worker(multiprocessing.Process):
def run(self):
print 'In %s' % self.name
return
if __name__ == '__main__':
jobs = []
for i in range(5):
p = Worker()
jobs.append(p)
p.start()
for j in jobs:
j.join()
Python中进程间传递消息的常用方式包括Queue和管道等机制。
import multiprocessing
class MyFancyClass(object):
def __init__(self, name):
self.name = name
def do_something(self):
proc_name = multiprocessing.current_process().name
print 'Doing something fancy in %s for %s!' % \
(proc_name, self.name)
def worker(q):
obj = q.get()
obj.do_something()
if __name__ == '__main__':
queue = multiprocessing.Queue()
p = multiprocessing.Process(target=worker, args=(queue,))
p.start()
queue.put(MyFancyClass('Fancy Dan'))
# Wait for the worker to finish
queue.close()
queue.join_thread()
p.join()
import multiprocessing
import time
class Consumer(multiprocessing.Process):
def __init__(self, task_queue, result_queue):
multiprocessing.Process.__init__(self)
self.task_queue = task_queue
self.result_queue = result_queue
def run(self):
proc_name = self.name
while True:
next_task = self.task_queue.get()
if next_task is None:
# Poison pill means shutdown
print '%s: Exiting' % proc_name
self.task_queue.task_done()
break
print '%s: %s' % (proc_name, next_task)
answer = next_task()
self.task_queue.task_done()
self.result_queue.put(answer)
return
class Task(object):
def __init__(self, a, b):
self.a = a
self.b = b
def __call__(self):
time.sleep(0.1) # pretend to take some time to do the work
return '%s * %s = %s' % (self.a, self.b, self.a * self.b)
def __str__(self):
return '%s * %s' % (self.a, self.b)
if __name__ == '__main__':
# Establish communication queues
tasks = multiprocessing.JoinableQueue()
results = multiprocessing.Queue()
# Start consumers
num_consumers = multiprocessing.cpu_count() * 2
print 'Creating %d consumers' % num_consumers
consumers = [ Consumer(tasks, results)
for i in xrange(num_consumers) ]
for w in consumers:
w.start()
# Enqueue jobs
num_jobs = 10
for i in xrange(num_jobs):
tasks.put(Task(i, i))
# Add a poison pill for each consumer
for i in xrange(num_consumers):
tasks.put(None)
# Wait for all of the tasks to finish
tasks.join()
# Start printing results
while num_jobs:
result = results.get()
print 'Result:', result
num_jobs -= 1
Event对象提供了一种简单的进程间状态同步机制。事件具有“设置”和“未设置”两种状态。用户可以通过指定可选的超时参数，等待事件状态从“未设置”变为“设置”，从而实现进程间的协调。
进程间信号传递（Event同步）的示例代码如下：
import multiprocessing
import time
def wait_for_event(e):
"""Wait for the event to be set before doing anything"""
print 'wait_for_event: starting'
e.wait()
print 'wait_for_event: e.is_set()->', e.is_set()
def wait_for_event_timeout(e, t):
"""Wait t seconds and then timeout"""
print 'wait_for_event_timeout: starting'
e.wait(t)
print 'wait_for_event_timeout: e.is_set()->', e.is_set()
if __name__ == '__main__':
e = multiprocessing.Event()
w1 = multiprocessing.Process(name='block',
target=wait_for_event,
args=(e,))
w1.start()
w2 = multiprocessing.Process(name='nonblock',
target=wait_for_event_timeout,
args=(e, 2))
w2.start()
print 'main: waiting before calling Event.set()'
time.sleep(3)
e.set()
print 'main: event is set'
在Python多进程编程中，Queue是进行进程间数据传递的常用且高效的方式。
Queue的基本使用示例如下：
from multiprocessing import Process, Queue def f(q): q.put([42, None, 'hello']) if __name__ == '__main__': q = Queue() p = Process(target=f, args=(q,)) p.start() print q.get() # prints "[42, None, 'hello']" p.join()
多线程环境中使用的优先队列Queue与多进程中的Queue在概念上类似，但实现机制不同，需注意区分。
import Queue
import threading
import time
exitFlag = 0
class myThread (threading.Thread):
def __init__(self, threadID, name, q):
threading.Thread.__init__(self)
self.threadID = threadID
self.name = name
self.q = q
def run(self):
print "Starting " + self.name
process_data(self.name, self.q)
print "Exiting " + self.name
def process_data(threadName, q):
while not exitFlag:
queueLock.acquire()
if not workQueue.empty():
data = q.get()
queueLock.release()
print "%s processing %s" % (threadName, data)
else:
queueLock.release()
time.sleep(1)
threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = Queue.Queue(10)
threads = []
threadID = 1
# Create new threads
for tName in threadList:
thread = myThread(threadID, tName, workQueue)
thread.start()
threads.append(thread)
threadID += 1
# Fill the queue
queueLock.acquire()
for word in nameList:
workQueue.put(word)
queueLock.release()
# Wait for queue to empty
while not workQueue.empty():
pass
# Notify threads it's time to exit
exitFlag = 1
# Wait for all threads to complete
for t in threads:
t.join()
print "Exiting Main Thread"
多进程通过Queue进行通信的完整示例如下：
import time
from multiprocessing import Process,Queue
MSG_QUEUE = Queue(5)
def startA(msgQueue):
while True:
if msgQueue.empty() > 0:
print ('queue is empty %d' % (msgQueue.qsize()))
else:
msg = msgQueue.get()
print( 'get msg %s' % (msg,))
time.sleep(1)
def startB(msgQueue):
while True:
msgQueue.put('hello world')
print( 'put hello world queue size is %d' % (msgQueue.qsize(),))
time.sleep(3)
if __name__ == '__main__':
processA = Process(target=startA,args=(MSG_QUEUE,))
processB = Process(target=startB,args=(MSG_QUEUE,))
processA.start()
print( 'processA start..')
在上述示例中，主进程定义了一个Queue类型的变量，并将其作为参数传递给子进程processA和processB。其中一个子进程负责向队列写入数据，另一个子进程负责从队列读取数据，从而实现进程间的数据交互。
更多关于Python相关技术的详细内容，欢迎查阅本站专题：《Python进程与线程操作技巧总结》、《Python Socket编程技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总》。
希望本文的讲解能对您的Python程序设计工作有所帮助。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
