python进程与线程

1. 前言        

        最近总结了python中如何创建并使用多进程和多线程的主要方法，所有代码都自测成功，用于以后编写代码时复制修改。

2. 各种代码

2.1 使用multiprocessing模块创建进程

        演示process类的方法和属性的使用，创建2个字进程，分别使用os模块和time模块输出父进程和子进程的ID以及子进程的时间，并调用process类的name和pid属性，代码如下：

# -*- coding:utf-8 -*-
from multiprocessing import Process
import time
import os


# 两个子进程将会调用的两个方法
def child_1(interval):
    print("子进程（%s）开始执行，父进程为（%s）" % (os.getpid(), os.getppid()))
    t_start = time.time()  # 计时开始
    time.sleep(interval)  # 程序将会被挂起interval秒
    t_end = time.time()  # 计时结束
    print("子进程（%s）执行时间为'%0.2f'秒" % (os.getpid(), t_end - t_start))


def child_2(interval):
    print("子进程（%s）开始执行，父进程为（%s）" % (os.getpid(), os.getppid()))
    t_start = time.time()  # 计时开始
    time.sleep(interval)  # 程序将会被挂起interval秒
    t_end = time.time()  # 计时结束
    print("子进程（%s）执行时间为'%0.2f'秒" % (os.getpid(), t_end - t_start))


if __name__ == '__main__':
    print("------父进程开始执行-------")
    print("父进程PID：%s" % os.getpid())  # 输出当前程序的ID
    p1 = Process(target=child_1, name="process1", args=(1,))  # 实例化进程p1
    p2 = Process(target=child_2, name="process2", args=(2,))  # 实例化进程p2
    p1.start()  # 启动进程p1
    p2.start()  # 启动进程p2
    # 同时父进程仍然往下执行，如果p2进程还在执行，将会返回True
    print("p1.is_alive=%s" % p1.is_alive())
    print("p2.is_alive=%s" % p2.is_alive())
    # 输出p1和p2进程的别名和PID
    print("p1.name=%s" % p1.name)
    print("p1.pid=%s" % p1.pid)
    print("p2.name=%s" % p2.name)
    print("p2.pid=%s" % p2.pid)
    print("------等待子进程-------")
    p1.join()  # 等待p1进程结束
    p2.join()  # 等待p2进程结束
    print("------父进程执行结束-------")

运行结果：

------父进程开始执行-------
父进程PID：616
p1.is_alive=True
p2.is_alive=True
p1.name=process1
p1.pid=12256
p2.name=process2
p2.pid=14976
------等待子进程-------
子进程（12256）开始执行，父进程为（616）
子进程（14976）开始执行，父进程为（616）
子进程（12256）执行时间为'1.00'秒
子进程（14976）执行时间为'2.00'秒
------父进程执行结束-------

Process finished with exit code 0

2.2 使用process子类创建进程

对于简单的任务，通常使用上一节方式实现多进程，但是如果要处理复杂任务，通常定义一个类，使其继承Process类，每次实例化这个类的时候，等同于实例化一个进程对象。代码如下：

# -*- coding:utf-8 -*-
from multiprocessing import Process
import time
import os


# 继承Process类
class SubProcess(Process):
    # 由于Process类本身也有__init__初识化方法，这个子类相当于重写了父类的这个方法
    def __init__(self, interval, name=''):
        Process.__init__(self)  # 调用Process父类的初始化方法
        self.interval = interval  # 接收参数interval
        if name:  # 判断传递的参数name是否存在
            self.name = name  # 如果传递参数name,则为子进程创建name属性，否则使用默认属性

    # 重写了Process类的run()方法
    def run(self):
        print("子进程(%s) 开始执行，父进程为（%s）" % (os.getpid(), os.getppid()))
        t_start = time.time()
        time.sleep(self.interval)
        t_stop = time.time()
        print("子进程(%s)执行结束，耗时%0.2f秒" % (os.getpid(), t_stop - t_start))


if __name__ == "__main__":
    print("------父进程开始执行-------")
    print("父进程PID：%s" % os.getpid())  # 输出当前程序的ID
    p1 = SubProcess(interval=1, name='subprocess1')
    p2 = SubProcess(interval=2, name='subprocess2')
    # 对一个不包含target属性的Process类执行start()方法，就会运行这个类中的run()方法，
    # 所以这里会执行p1.run()
    p1.start()  # 启动进程p1
    p2.start()  # 启动进程p2
    # 输出p1和p2进程的执行状态，如果真正进行，返回True,否则返回False
    print("p1.is_alive=%s" % p1.is_alive())
    print("p2.is_alive=%s" % p2.is_alive())
    # 输出p1和p2进程的别名和PID
    print("p1.name=%s" % p1.name)
    print("p1.pid=%s" % p1.pid)
    print("p2.name=%s" % p2.name)
    print("p2.pid=%s" % p2.pid)
    print("------等待子进程-------")
    p1.join()  # 等待p1进程结束
    p2.join()  # 等待p2进程结束
    print("------父进程执行结束-------")

运行结果：

------父进程开始执行-------
父进程PID：8292
p1.is_alive=True
p2.is_alive=True
p1.name=subprocess1
p1.pid=7376
p2.name=subprocess2
p2.pid=8904
------等待子进程-------
子进程(8904) 开始执行，父进程为（8292）子进程(7376) 开始执行，父进程为（8292）

子进程(7376)执行结束，耗时1.00秒
子进程(8904)执行结束，耗时2.02秒
------父进程执行结束-------

Process finished with exit code 0

2.3 使用进程池Pool创建进程

定义一个进程池，设置最大进程数为3，然后使用非阻塞方式执行10个任务，查看每个进程执行的任务，代码如下：

# -*- coding=utf-8 -*-
from multiprocessing import Pool
import os, time


def task(name):
    print('子进程（%s）执行task %s ...' % (os.getpid(), name))
    time.sleep(1)  # 休眠1秒


if __name__ == '__main__':
    print('父进程（%s）.' % os.getpid())
    p = Pool(3)  # 定义一个进程池，最大进程数3
    for i in range(10):  # 从0开始循环10次
        p.apply_async(task, args=(i,))  # 使用非阻塞方式调用task()函数
    print('等待所有子进程结束...')
    p.close()  # 关闭进程池，关闭后p不再接收新的请求
    p.join()  # 等待子进程结束
    print('所有子进程结束.')

执行结果：

父进程（2992）.
等待所有子进程结束...
子进程（3772）执行task 0 ...
子进程（11708）执行task 1 ...
子进程（5084）执行task 2 ...
子进程（11708）执行task 3 ...
子进程（5084）执行task 4 ...
子进程（3772）执行task 5 ...
子进程（3772）执行task 6 ...
子进程（5084）执行task 7 ...
子进程（11708）执行task 8 ...
子进程（3772）执行task 9 ...
所有子进程结束.

Process finished with exit code 0

2.4 队列的基本功能

代码如下：

# coding=utf-8
from multiprocessing import Queue

if __name__ == '__main__':
    q = Queue(3)  # 初始化一个Queue对象，最多可接收三条put消息
    q.put("消息1")
    q.put("消息2")
    print(q.full())  # 返回False
    q.put("消息3")
    print(q.full())  # 返回True

    # 因为消息列队已满，下面的try都会抛出异常，
    # 第一个try会等待2秒后再抛出异常，第二个try会立刻抛出异常
    try:
        q.put("消息4", True, 2)
    except:
        print("消息列队已满，现有消息数量:%s" % q.qsize())

    try:
        q.put_nowait("消息4")
    except:
        print("消息列队已满，现有消息数量:%s" % q.qsize())

    # 读取消息时，先判断消息列队是否为空，再读取
    if not q.empty():
        print('----从队列中获取消息---')
        for i in range(q.qsize()):
            print(q.get_nowait())
    # 先判断消息列队是否已满，再写入
    if not q.full():
        q.put_nowait("消息4")

执行结果：

False
True
消息列队已满，现有消息数量:3
消息列队已满，现有消息数量:3
----从队列中获取消息---
消息1
消息2
消息3

Process finished with exit code 0

2.5 使用队列再进程间通信

创建两个子进程，一个子进程负责向队列中写入数据，另外一个子进程负责从队列中读取数据。代码如下：

# -*- coding: utf-8 -*-
from multiprocessing import Process, Queue
import time


# 向队列中写入数据
def write_task(q):
    if not q.full():
        for i in range(5):
            message = "消息" + str(i)
            q.put(message)
            print("写入:%s" % message)


# 从队列读取数据
def read_task(q):
    time.sleep(1)  # 休眠1秒
    while not q.empty():
        print("读取:%s" % q.get(True, 2))  # 等待2秒，如果还没读取到任何消息，
    # 则抛出"Queue.Empty"异常


if __name__ == "__main__":
    print("-----父进程开始-----")
    q = Queue()  # 父进程创建Queue，并传给各个子进程
    pw = Process(target=write_task, args=(q,))  # 实例化写入队列的子进程，并且传递队列
    pr = Process(target=read_task, args=(q,))  # 实例化读取队列的子进程，并且传递队列
    pw.start()  # 启动子进程 pw，写入
    pr.start()  # 启动子进程 pr，读取
    pw.join()  # 等待 pw 结束
    pr.join()  # 等待 pr 结束
    print("-----父进程结束-----")

执行结果：

-----父进程开始-----
写入:消息0
写入:消息1
写入:消息2
写入:消息3
写入:消息4
读取:消息0
读取:消息1
读取:消息2
读取:消息3
读取:消息4
-----父进程结束-----

Process finished with exit code 0

2.6 使用theading模块创建线程

使用theading模块创建线程，代码如下：

# -*- coding:utf-8 -*-
import threading, time


def process():
    for i in range(3):
        time.sleep(1)
        print("thread name is %s" % threading.current_thread().name)


if __name__ == '__main__':
    print("-----主线程开始-----")
    threads = [threading.Thread(target=process) for i in range(4)]  # 创建4个线程，存入列表
    for t in threads:
        t.start()  # 开启线程
    for t in threads:
        t.join()  # 等待子线程结束
    print("-----主线程结束-----")

执行结果：

-----主线程开始-----
thread name is Thread-2
thread name is Thread-4
thread name is Thread-3
thread name is Thread-1
thread name is Thread-3
thread name is Thread-1
thread name is Thread-4
thread name is Thread-2
thread name is Thread-2thread name is Thread-3
thread name is Thread-1thread name is Thread-4


-----主线程结束-----

Process finished with exit code 0

2.7 使用Thead子类创建线程

创建一个子类SubThread，继承threading.Thread线程类,并定义一个run()方法。实例化SubThread类创建2个线程，并且调用start()方法开启线程，程序会自动调用run()方法。代码如下：

# -*- coding: utf-8 -*-
import threading
import time


class SubThread(threading.Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            msg = "子线程" + self.name + '执行，i=' + str(i)  # name属性中保存的是当前线程的名字
            print(msg)


if __name__ == '__main__':
    print('-----主线程开始-----')
    t1 = SubThread()  # 创建子线程t1
    t2 = SubThread()  # 创建子线程t2
    t1.start()  # 启动子线程t1
    t2.start()  # 启动子线程t2
    t1.join()  # 等待子线程t1
    t2.join()  # 等待子线程t2
    print('-----主线程结束-----')

执行结果：

-----主线程开始-----
子线程Thread-2执行，i=0
子线程Thread-1执行，i=0
子线程Thread-1执行，i=1
子线程Thread-2执行，i=1
子线程Thread-2执行，i=2子线程Thread-1执行，i=2

-----主线程结束-----

Process finished with exit code 0

2.8 线程间通信

进程间不能共享信息，但是线程间可以通过全局变量共享信息。代码如下

# -*- coding:utf-8 -*-
from threading import Thread
import time


def plus():
    print('-------子线程1开始------')
    global g_num
    g_num += 50
    print('g_num is %d' % g_num)
    print('-------子线程1结束------')


def minus():
    time.sleep(1)
    print('-------子线程2开始------')
    global g_num
    g_num -= 50
    print('g_num is %d' % g_num)
    print('-------子线程2结束------')


g_num = 100  # 定义一个全局变量
if __name__ == '__main__':
    print('-------主线程开始------')
    print('g_num is %d' % g_num)
    t1 = Thread(target=plus)  # 实例化线程p1
    t2 = Thread(target=minus)  # 实例化线程p2
    t1.start()  # 开启线程p1
    t2.start()  # 开启线程p2
    t1.join()  # 等待p1线程结束
    t2.join()  # 等待p2线程结束
    print('-------主线程结束------')

执行结果：

-------主线程开始------
g_num is 100
-------子线程1开始------
g_num is 150
-------子线程1结束------
-------子线程2开始------
g_num is 100
-------子线程2结束------
-------主线程结束------

Process finished with exit code 0

2.9 使用互斥锁

互斥锁可以防止多个线程同事读写某一块内存区域。代码如下：

from threading import Thread, Lock
import time

n = 100  # 共100张票


def task():
    global n
    mutex.acquire()  # 上锁
    temp = n  # 赋值给临时变量
    time.sleep(0.1)  # 休眠0.1秒
    n = temp - 1  # 数量减1
    print('购买成功，剩余%d张电影票' % n)
    mutex.release()  # 释放锁


if __name__ == '__main__':
    mutex = Lock()  # 实例化Lock类
    t_l = []  # 初始化一个列表
    for i in range(10):
        t = Thread(target=task)  # 实例化线程类
        t_l.append(t)  # 将线程实例存入列表中
        t.start()  # 创建线程
    for t in t_l:
        t.join()  # 等待子线程结束

执行结果：

购买成功，剩余99张电影票
购买成功，剩余98张电影票
购买成功，剩余97张电影票
购买成功，剩余96张电影票
购买成功，剩余95张电影票
购买成功，剩余94张电影票
购买成功，剩余93张电影票
购买成功，剩余92张电影票
购买成功，剩余91张电影票
购买成功，剩余90张电影票

Process finished with exit code 0

2.10 使用队列在线程间通信

进程间通信可以用multiprocessing模块的Queue队列，线程间也可以使用Queue队列实现线程间通信。使用Queue在线程间通信通常应用于生产者消费者模式。代码如下：

from queue import Queue
import random, threading, time


# 生产者类
class Producer(threading.Thread):
    def __init__(self, name, queue):
        threading.Thread.__init__(self, name=name)
        self.data = queue

    def run(self):
        for i in range(5):
            print("生成者%s将产品%d加入队列!" % (self.getName(), i))
            self.data.put(i)
            time.sleep(random.random())
        print("生成者%s完成!" % self.getName())


# 消费者类
class Consumer(threading.Thread):
    def __init__(self, name, queue):
        threading.Thread.__init__(self, name=name)
        self.data = queue

    def run(self):
        for i in range(5):
            val = self.data.get()
            print("消费者%s将产品%d从队列中取出!" % (self.getName(), val))
            time.sleep(random.random())
        print("消费者%s完成!" % self.getName())


if __name__ == '__main__':
    print('-----主线程开始-----')
    queue = Queue()  # 实例化队列
    producer = Producer('Producer', queue)  # 实例化线程Producer，并传入队列作为参数
    consumer = Consumer('Consumer', queue)  # 实例化线程Consumer，并传入队列作为参数
    producer.start()  # 启动线程Producer
    consumer.start()  # 启动线程Consumer
    producer.join()  # 等待线程Producer结束
    consumer.join()  # 等待线程Consumer结束
    print('-----主线程结束-----')

执行结果：

-----主线程开始-----
生成者Producer将产品0加入队列!
消费者Consumer将产品0从队列中取出!
生成者Producer将产品1加入队列!
消费者Consumer将产品1从队列中取出!
生成者Producer将产品2加入队列!
消费者Consumer将产品2从队列中取出!
生成者Producer将产品3加入队列!
消费者Consumer将产品3从队列中取出!
生成者Producer将产品4加入队列!
生成者Producer完成!
消费者Consumer将产品4从队列中取出!
消费者Consumer完成!
-----主线程结束-----

Process finished with exit code 0

3.最后

这是我总结的python代码编写常的有关进程线程的代码。基本功能应该不外呼这10组代码。