进程and线程

在聊进程和线程之前,先说说多任务,什么叫“多任务”呢？简单地说，就是操作系统可以同时运行多个任务。

进程 / 程序

• 编写完毕的代码，在没有运行的时候，称之为程序.
• 正在运行着的代码，就成为进程.
• 进程，除了包含代码以外，还有需要运行的环境等，所以和程序是有区别的.

fork():

Python的os模块封装了常见的系统调用，其中就包括fork，可以在Python程序中轻松创建子进程：

import os
pid = os.fork()
print(pid)  # 父进程返回子进程号,子进程返回0
if pid == 0:
    print('子进程',os.getpid(),os.getppid()) # 获取进程号和父进程号
else:
    print('父进程',os.getpid())

1. 程序执行到os.fork()时，操作系统会创建一个新的进程（子进程），然后复制父进程的所有信息到子进程中.
2. 然后父进程和子进程都会从fork()函数中得到一个返回值，在子进程中这个值一定是0，而父进程中是子进程的 id号.

• 说到了fork(),它在windows上是不可以运行的.

在window下没有一个函数可以实现UNIX下的fork()函数,其原因是历史造成的.对于UNIX来说它一出生就是多用户的系统,所以它的所有进程都共有一个最原始的父进程init.而windows生下来时是个单用户系统(DOS),不存在这样的概念.所以fork这个函数是UNIX下特有的.

在Unix/Linux操作系统中，提供了一个fork()系统函数，它非常特殊。
普通的函数调用，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（称为父进程）复制了一份（称为子进程），然后，分别在父进程和子进程内返回。
子进程永远返回0，而父进程返回子进程的ID。
这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。

getpid() / getppid()

import os

pid = os.fork()
if pid<0:
    print("fork调用失败。")
elif pid == 0:
    print("我是子进程（%s），我的父进程是（%s）"%(os.getpid(),os.getppid()))
    x+=1
else:
    print("我是父进程（%s），我的子进程是（%s）"%(os.getpid(),pid))

print("父子进程都可以执行这里的代码")

运行结果如下:

我是父进程（19360），我的子进程是（19361）
父子进程都可以执行这里的代码
我是子进程（19361），我的父进程是（19360）
父子进程都可以执行这里的代码

创建进程:

multiprocessing模块提供了一个Process类来代表一个进程对象

• 在windows中运行进程的python代码时,需要在主程序之上加上一句:

if __name__ == '__main__':    # 将调用写在这个if内.

from multiprocessing import Process
import os

# 子进程要执行的代码
def run_proc(name):
    print('子进程运行中，name= %s ,pid=%d...' % (name, os.getpid()))

if __name__=='__main__':
    print('父进程 %d.' % os.getpid())
    p = Process(target=run_proc, args=('test',))
    print('子进程将要执行')
    p.start()
    p.join()
    print('子进程已结束')

创建子进程时，只需要传入一个执行函数和函数的参数，创建一个Process实例，用start()方法启动，这样创建进程比fork()还要简单。

Process语法结构如下：

Process([group [, target [, name [, args [, kwargs]]]]])

• target：表示这个进程实例所调用对象；
• args：表示调用对象的位置参数元组；
• kwargs：表示调用对象的关键字参数字典；
• name：为当前进程实例的别名；
• group：大多数情况下用不到；

Process类常用方法：

• is_alive()：判断进程实例是否还在执行；
• join([timeout])：是否等待进程实例执行结束，或等待多少秒；
• start()：启动进程实例（创建子进程）；
• run()：如果没有给定target参数，对这个对象调用start()方法时，就将执行对象中的run()方法；
• terminate()：不管任务是否完成，立即终止；

Process类常用属性：

• name：当前进程实例别名，默认为Process-N，N为从1开始递增的整数；
• pid：当前进程实例的PID值；

创建新的进程还能够使用类的方式，可以自定义一个类，继承Process类，每次实例化这个类的时候，就等同于实例化一个进程对象.

import time
from multiprocessing import Process
class MyProcess(Process):
    def __init__(self):
        super().__init__()  # 先初始化父类

    def run(self): # 重写run方法
        for i in range(5):
            time.sleep(1)
            print('哈哈')

p = MyProcess()
p.start()
p.join(3)
print('呵呵')

进程池:

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来执行.

from multiprocessing import Pool
import time
def work():
    for i in range(5):
        time.sleep(1)
        print('老王')

p = Pool(3)
for i in range(3):
    # p.apply_async(work)  # 非阻塞添加进程
    p.apply(work)    # 阻塞添加进程
    print('添加成功')

p.close()  # 添加完成关闭池子
p.join()

• 池子必须关闭

multiprocessing.Pool常用函数解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
• apply(func[, args[, kwds]])：使用阻塞方式调用func
• close()：关闭Pool，使其不再接受新的任务；
• terminate()：不管任务是否完成，立即终止；
• join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

进程间的通信:

Process之间有时需要通信，操作系统提供了很多机制来实现进程间的通信。

Queue的使用:

可以使用multiprocessing模块的Queue实现多进程之间的数据传递，Queue本身是一个消息列队程序，首先用一个小实例来演示一下Queue的工作原理：

from multiprocessing import Queue
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.full():
    q.put_nowait("消息4")

#读取消息时，先判断消息列队是否为空，再读取
if not q.empty():
    for i in range(q.qsize()):
        print(q.get_nowait())

初始化Queue()对象时（例如：q=Queue()），若括号中没有指定最大可接收的消息数量，或数量为负值，那么就代表可接受的消息数量没有上限（直到内存的尽头）；

• Queue.qsize()：返回当前队列包含的消息数量；
• Queue.empty()：如果队列为空，返回True，反之False ；
• Queue.full()：如果队列满了，返回True,反之False；
• Queue.get([block[, timeout]])：获取队列中的一条消息，然后将其从列队中移除，block默认值为True；

进程池中的Queue:

from multiprocessing import Manager,Pool
import time
def write(q):
    for i in range(10):
        time.sleep(0.5)
        print('添加成功')
        q.put(i)

def read(q):
    while True:
        if q.qsize()>0:
            for i in range(q.qsize()):
                a = q.get()
                print(a)
            if a == 9:
                break


p = Pool(3)
q = Manager().Queue()
p.apply_async(write,(q,))
p.apply_async(read,(q,))
p.close()
p.join()

线程:

python的thread模块是比较底层的模块，python的threading模块是对thread做了一些包装的，可以更加方便的被使用:

使用threading模块

import threading
import time

def saySorry():
    print("亲爱的，我错了，我能吃饭了吗？")
    time.sleep(1)

if __name__ == "__main__":
    for i in range(5):
        t = threading.Thread(target=saySorry)
        t.start() #启动线程，即让线程开始执行

• 可以明显看出使用了多线程并发的操作，花费时间要短很多
• 创建好的线程，需要调用start()方法来启动
• 主线程会等待所有的子线程结束后才结束

利用面向对象实现:

from threading import Thread
import time

class MyThread(Thread):
    def run(self):
        for i in range(3):
            time.sleep(1)
            print('我爱你')

t = MyThread()
t.start()

• 查看线程数量:

length = len(threading.enumerate())
print('当前运行的线程数为：%d'%length)

在一个进程内的所有线程共享全局变量，能够在不适用其他方式的前提下完成多线程之间的数据共享（这点要比多进程要好）

缺点就是，线程是对全局变量随意遂改可能造成多线程之间对全局变量的混乱（即线程非安全）

from threading import Thread
import threading,time
# 多进程修改全局变量是非安全的.
num = 0
flag = True

def test1():
    global num
    global flag
    if flag:
        for i in range(1000000):
            num += 1
        flag = False
    print(num)
    

def test2():
    global num
    global flag
    while True:
        if not flag:
            for i in range(1000000):
                num += 1
            flag = True
            break
    print(num)



t1 = Thread(target=test1)
t1.start()
# time.sleep(2)
t2 = Thread(target=test2)
t2.start()

进程 / 线程

进程是系统进行资源分配和调度的一个独立单位.
线程是进程的一个实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位.

线程和进程在使用上各有优缺点：线程执行开销小，但不利于资源的管理和保护；而进程正相反。

互斥锁

当多个线程几乎同时修改某一个共享数据的时候，需要进行同步控制

• 线程同步能够保证多个线程安全访问竞争资源，最简单的同步机制是引入互斥锁。
• 互斥锁为资源引入一个状态：锁定/非锁定。

threading模块中定义了Lock类，可以方便的处理锁定：

#创建锁
mutex = threading.Lock()
#锁定
mutex.acquire([blocking])
#释放
mutex.release()

• 但是这样做的话,多进程就没有意义了,这样只有单进程.

死锁:

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。
尽管死锁很少发生，但一旦发生就会造成应用的停止响应。下面看一个死锁的例子

import threading
import time

class MyThread1(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name+'----do1---up----')
            time.sleep(1)

            if mutexB.acquire():
                print(self.name+'----do1---down----')
                mutexB.release()
            mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name+'----do2---up----')
            time.sleep(1)
            if mutexA.acquire():
                print(self.name+'----do2---down----')
                mutexA.release()
            mutexB.release()

mutexA = threading.Lock()
mutexB = threading.Lock()

if __name__ == '__main__':
    t1 = MyThread1()
    t2 = MyThread2()
    t1.start()
    t2.start()

银行家算法:

[背景知识]
一个银行家如何将一定数目的资金安全地借给若干个客户，使这些客户既能借到钱完成要干的事，同时银行家又能收回全部资金而不至于破产，这就是银行家问题。这个问题同操作系统中资源分配问题十分相似：银行家就像一个操作系统，客户就像运行的进程，银行家的资金就是系统的资源。

[问题的描述]
一个银行家拥有一定数量的资金，有若干个客户要贷款。每个客户须在一开始就声明他所需贷款的总额。若该客户贷款总额不超过银行家的资金总数，银行家可以接收客户的要求。客户贷款是以每次一个资金单位（如1万RMB等）的方式进行的，客户在借满所需的全部单位款额之前可能会等待，但银行家须保证这种等待是有限的，可完成的。

例如：有三个客户C1，C2，C3，向银行家借款，该银行家的资金总额为10个资金单位，其中C1客户要借9各资金单位，C2客户要借3个资金单位，C3客户要借8个资金单位，总计20个资金单位。

对于a图的状态，按照安全序列的要求，我们选的第一个客户应满足该客户所需的贷款小于等于银行家当前所剩余的钱款，可以看出只有C2客户能被满足：C2客户需1个资金单位，小银行家手中的2个资金单位，于是银行家把1个资金单位借给C2客户，使之完成工作并归还所借的3个资金单位的钱，进入b图。同理，银行家把4个资金单位借给C3客户，使其完成工作，在c图中，只剩一个客户C1，它需7个资金单位，这时银行家有8个资金单位，所以C1也能顺利借到钱并完成工作。最后（见图d）银行家收回全部10个资金单位，保证不赔本。那麽客户序列{C1，C2，C3}就是个安全序列，按照这个序列贷款，银行家才是安全的。否则的话，若在图b状态时，银行家把手中的4个资金单位借给了C1，则出现不安全状态：这时C1，C3均不能完成工作，而银行家手中又没有钱了，系统陷入僵持局面，银行家也不能收回投资。

综上所述，银行家算法是从当前状态出发，逐个按安全序列检查各客户谁能完成其工作，然后假定其完成工作且归还全部贷款，再进而检查下一个能完成工作的客户，......。如果所有客户都能完成工作，则找到一个安全序列，银行家才是安全的。

同步执行:

from threading import Thread,Lock
from time import sleep

class Task1(Thread):
    def run(self):
        while True:
            if lock1.acquire():
                print("------Task 1 -----")
                sleep(0.5)
                lock2.release()

class Task2(Thread):
    def run(self):
        while True:
            if lock2.acquire():
                print("------Task 2 -----")
                sleep(0.5)
                lock3.release()

class Task3(Thread):
    def run(self):
        while True:
            if lock3.acquire():
                print("------Task 3 -----")
                sleep(0.5)
                lock1.release()

#使用Lock创建出的锁默认没有“锁上”
lock1 = Lock()
#创建另外一把锁，并且“锁上”
lock2 = Lock()
lock2.acquire()
#创建另外一把锁，并且“锁上”
lock3 = Lock()
lock3.acquire()

t1 = Task1()
t2 = Task2()
t3 = Task3()

t1.start()
t2.start()
t3.start()

使用ThreadLocal的方法

ThreadLocal应运而生，不用查找dict，ThreadLocal帮你自动做这件事：

import threading

# 创建全局ThreadLocal对象:
local_school = threading.local()

def process_student():
    # 获取当前线程关联的student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))

def process_thread(name):
    # 绑定ThreadLocal的student:
    local_school.student = name
    process_student()

t1 = threading.Thread(target= process_thread, args=(‘laozhao',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('xiaoyuan',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()