fork

windows中是没有fork函数的，一开始直接在Windows中测试，直接报错

import os
import time

ret = os.fork()
if ret == 0:
    while True:
        print("-----1----")
        time.sleep(1)
else:
    while True:
        print("-----2-----")
        time.sleep(1)

AttributeError: module 'os' has no attribute 'fork'

然后在Linux上实验，可以。父子进程执行顺序是不定的。

fork函数调动一次，有两个返回值，父进程返回值>0，子进程返回值=0；

import os
ret = os.fork()
print(ret)

父进程中fork的返回值就是创建的子进程的pid。

import os
ret = os.fork()
print(ret)
if ret>0:
    print("--父进程pid--%d"%os.getpid())
else:
    print("--子进程pid--%d--父进程---%d"%(os.getpid(),os.getppid()))

父子进程的退出

父进程可以在子进行前退出。

import os
import time

ret = os.fork()

if ret==1:
    print("子进程1")
    time.sleep(5)
    print("子进程2")
else:
    print("父进程")
    time.sleep(3)

print("over")

全局变量问题：

变量在父子进程间不会共享，子进程会自己创建一份。

import os
import time

num = 100
ret = os.fork()
if ret==0:
    print("----process1----")
    num += 1
    print("-----process1---num=%d"%(num))

else:
    time.sleep(3)
    print("---process2----num=%d"%(num))

多次fork问题:

import os
import time

ret = os.fork()
if ret==0:
    print("--1--")
else:
    print("--2--")
ret = os.fork()
if ret ==0:
    print("--11--")
else:
    print("--22--")

3个fork问题：

Process创建子进程

multiprocessing模块中Process类

rom multiprocessing import Process
import time

def test():
    while True:
        print("--test--")
        time.sleep(1)

p = Process(target=test)//创建一个进程对象，test函数结束了，这个进程就结束了
p.start()

while True:
    print("--main()--")
    time.sleep(1)
                    

主进程等待子进程先结束

from multiprocessing import Process
import time

def test():
    while True:
        print("--test--")
        time.sleep(1)

p = Process(target=test)
p.start()
#这里主进程不会直接结束，它会等待子进程结束了再结束

Process的init函数

def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):

给target函数传递参数

args：位置参数元祖

kwargs：关键字参数字典

from multiprocessing import Process
import os

def test(name1,name2):
    print("name1=%s,name2=%s"%(name1,name2))

p=Process(target=test,args=("hy","ym"))

p.start()

join阻塞

join([timeout])：等待子进程结束，或等待多少秒（等待时间到了，就不等了）

terminate()：不管任务是否完成，立即终止进程

from multiprocessing import Process
import time

def test():
    for i in range(5):
        print("-%d--"%(i))
        time.sleep(1)

p = Process(target=test)
p.start()

p.join()#阻塞，等待子进程结束

print("--main--")

Process子类创建进程

from multiprocessing import Process
import time

class MyProcess(Process):
    def run(self):
        while True:
            print("--1--")
            time.sleep(1)

p = MyProcess()
p.start()#当没有给process传递target时，会自动执行run函数

while True:
    print("--main--")
    time.sleep(1)
                      

对一个不包含target属性的Process类执行start方法，会运行这个类中的run方法。

进程池Pool

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

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

from multiprocessing import Pool
import time
import os

def Worker(msg):
    print("%d,%d进程正在执行"%(msg,os.getpid()))
    time.sleep(3)


po=Pool(3)#3表示，进程池最多有3个进程一起执行
for i in range(10):
    #向进程添加任务，如果任务超过进程池中进程个数，需要等待，有任务执行完成，会继续执行。
    po.apply_async(Worker,(i,))


po.close()#关闭进程池，相当于不能添加新任务了
po.join()#主进程默认不会等待进程池中任务执行完成才推出，所以这里要阻塞，不然主进程结束了，进程池中任务没有执行完成也会跟着结束。
          

上面的代码在Windows中运行时，需要加一个

if __name__ == "__main__":

不然会报错。

pool中apply阻塞方式

from multiprocessing import Pool
import time
import os

def Worker(msg):
    print("%d,%d进程正在执行"%(msg,os.getpid()))
    time.sleep(3)


po=Pool(3)
for i in range(10):
    print(i)
    po.apply(Worker,(i,))#主进程在这里阻塞，等待一个任务的完成


po.close()
po.join()
          

进程间通信 :

Queue

In [1]: from multiprocessing import Queue

In [2]: q = Queue(3)//队列最大长度为3，只能放三个数据，存放类型随意

In [3]: q.empty()
Out[3]: True

In [4]: q.full()
Out[4]: False

In [5]: q.put("hy")

In [6]: q.put(123)

In [7]: q.get()
Out[7]: 'hy'

In [8]: q.get()
Out[8]: 123

In [9]: q.qsize()
Out[9]: 0

from multiprocessing import Queue,Process
import time

def write(q):
    for value in ["A","B","c"]:
        print("将数据%s放入队列"%value)
        q.put(value)
        time.sleep(1)


def read(q):
    while True:
        if not q.empty():
            value=q.get();
            print("将数据%s从队列中取出"%value)
            time.sleep(1)
        else:
            break;

if __name__ == "__main__":
    q=Queue()#这里没有指明队列长度，表示长度无限
    pw = Process(target=write,args=(q,))
    pr = Process(target=read,args=(q,))
    pw.start()
    pw.join()
    pr.start()
    pr.join()

    print("所有数据以及写读完成")

当使用pool时需要用Manager中的Queue

from multiprocessing import Manager,Pool
import os

def write(q):
    print("wirte启动(%d),其父进程为(%d)"%(os.getpid(),os.getppid()))
    for i in "huangtieniu":
        q.put(i)


def read(q):
    print("read进程启动(%d),其父进程为(%d)"%(os.getpid(),os.getppid()))
    for i in range(q.qsize()):
        print("read从进程中获得消息:%s"%q.get())


if __name__ == "__main__":
    q = Manager().Queue()
    po = Pool()
    po.apply(write,(q,))
    po.apply(read,(q,))
    po.close()
    po.join()
    print("End")

线程：

Python中thread模块是比较底层的模块，Python中的threading模块是对thread做了一层封装的，可以更加方便的使用。

主线程会等待子线程执行完毕再结束。是为了回收子线程的资源

from threading import Thread
import time

#多个线程执行的是同一个函数，相互之间不影响
def test():
    print("I'm so sorry.")
    time.sleep(1)


for i in range(5):
    t = Thread(target=test)
    t.start()

使用Thread子类创建多线程

import threading
import time

class MyThread(threading.Thread):
    def run(self):
        for i in range(3):
            print("I'm %s @ %d"%(self.name,i))
            time.sleep(1)


if __name__ == "__main__":
    t = MyThread()
    t.start()

线程的执行顺序也是不固定的，由操作系统说了算。

import threading
import time

def test1():
    for i in range(3):
        print("----%d------"%i)
        time.sleep(1)

def test2():
    for i in range(4,7):
        print("-----%d-----"%i)
        time.sleep(1)

if __name__ == "__main__":
    t1 = threading.Thread(target=test1)
    t2 = threading.Thread(target=test2)
    t1.start()
    t2.start()

线程共享全局变量，可以利用全局变量通信

from threading import Thread
import time

def work1():
    global num
    num +=1
    print("work1中num=%d"%(num))

def work2():
    global num
    print("work2中num=%d"%(num))

#线程之间会共享全局变量
num = 100

if __name__ == "__main__":
    t1 = Thread(target=work1)
    t2 = Thread(target=work2)
    t1.start()
    time.sleep(1)
    t2.start()

线程共享全局变量可能的问题：

from threading import Thread
import time

def test1():
    global num
    for i in range(1000000):
        num += 1
    print("--test1--num=%d"%(num))

def test2():
    global num
    for i in range(1000000):
        num += 1
    print("--test2--num=%d"%(num))


num = 0

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

time.sleep(3)这一行注释掉时结果是前一个，加上时结果是后一个。

列表当做实参传递给线程

import threading import Thread
import time

def work1(num):
    num.append(44)
    print("--in work1--",num)

def work2(num):
    time.sleep(1)#保证work1执行完成
    print("--in work2--",num)

num = [11,22,33]

if __name__ == "__main__":
    t1 = Thread(target=work1,args=(num,))
    t2 = Thread(target=work2,args=(num,))
    t1.start()
    t2.start()

避免全局变量被修改的方式

1.加一个flag标识，轮训

from threading import Thread
import time



def test1():
    global num
    global flag
    if flag == 0:
        for i in range(1000000):
            num += 1
        flag = 1
    print("--test1--num=%d"%(num))

def test2():
    global num
    #轮训，太耗费cpu
    while True:
        if flag != 0:
            for i in range(1000000):
                num += 1
            break
    print("--test2--num=%d"%(num))


num = 0
flag = 0

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    #time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

避免全局变量被修改---互斥锁

某个线程要更改共享数据时，先将其锁定，此时资源的状态为“锁定”，其他线程不能更改；知道该线程释放资源，将资源的状态变为“非锁定”，其他的线程才能再次锁定该资源。互斥锁保证了每次只有一个线程进行写入操作，从而保证了多线程情况下数据的正确性。

from threading import Thread,Lock
import time



def test1():
    global num
    #加锁，test1和test2竞争加锁，如果有一方加锁，另一方会阻塞，知道这个所被解开
    mutex.acquire()
    for i in range(1000000):
        num += 1
    #解锁
    mutex.release()
    print("--test1--num=%d"%(num))

def test2():
    global num
    mutex.acquire()
    for i in range(1000000):
        num += 1
    mutex.release()
    print("--test2--num=%d"%(num))


num = 0
flag = 0

#创建一把互斥锁，默认是没有上锁的
mutex = Lock()

if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test2)
    p1.start()
    #time.sleep(3)
    p2.start()

    print("--主线程中num=%d"%(num))

加锁原则：

能不加锁就不加锁，加锁的范围能满足要求就好，不要太大。等待解锁的方式：通知，不是轮训。

多个线程使用非全局变量，非全局变量各自是各自的，不需要加锁

from threading import Thread
import time
import threading


def test1():
    name = threading.current_thread().name
    print("thread name is %s"%(name))
    num = 100
    if name == "Thread-1":
        num += 1
    else:
        time.sleep(2)
    print("thread is %s,num is %d"%(name,num))



if __name__ == "__main__":
    p1 = Thread(target=test1)
    p2 = Thread(target=test1)
    p1.start()
    p2.start()

死锁

在线程间共享多个资源的时候，如果两个线程分别占有一部分资源并且同时等待对方的资源，就会造成死锁。

import time
import threading
from threading import Lock

class MyThread1(threading.Thread):
    def run(self):
        if mutexA.acquire():
            print(self.name+"已对A加锁")
            time.sleep(1)

            if mutexB.acquire():
                print(self.name+"已对B加锁")
            mutexB.release()
        mutexA.release()

class MyThread2(threading.Thread):
    def run(self):
        if mutexB.acquire():
            print(self.name+"已对B加锁")
            time.sleep(1)

            if mutexA.acquire():
                print(self.name+"已对A加锁")
            mutexA.release()
        mutexB.release()

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

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