多任务的线程与进程

多任务

• 简单地说，就是操作系统可以同时运行多个任务。打个比方，你一边在用浏览器上网，一边在听MP3，一边在用Word赶作业，这就是多任务，至少同时有3个任务正在运行。

进程和程序

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

python 操作多进程（fork（））

• 父进程、子进程执行顺序没有规律，完全取决于操作系统的调度算法

import os
# fork()函数只能在ubuntu/linux/mac运行
pid = os.fork()
if pid == 0:
    print('子进程')
else:
    print('父进程')

getpid()、getppid()

• 程序执行到os.fork()时，操作系统会创建一个新的进程（子进程），然后复制父进程的所有信息到子进程中。然后父进程和子进程都会从fork()函数中得到一个返回值，在子进程中这个值一定是0，而父进程中是子进程的 id号
• 子进程永远返回0，而父进程返回子进程的ID。
• 一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getppid()就可以拿到父进程的ID。

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

多进程修改全局变量

import os
import time
num = 0
pid = os.fork()
if pid == 0:
    time.sleep(5)
    print("子进程",num)
    if num == 1:
        print("msg")
else:
    time.sleep(1)
    num+=1
print("父进程",num)

多次fork问题（不要使用while True）

import os.fork
os.fork()
os.fork()
os.fork()
#fork炸弹
'''
while True:
    os.fork()
'''

多进程中，每个进程中所有数据（包括全局变量）都各有拥有一份，互不影响

python跨平台多进程模块（multiprocessing）

from multiprocessing import Process
import time
def work(num):
    print(num)
    for i in range(5):
        time.sleep(1)
        print("呵呵")
p = Process(target = work,args=(1,)) # 创建子进程
p.start()
p.join(3)#等子进程结束后 在执行主进程代码 3指的是超时时间
print("哈哈")

面向对象实现多进程

from multiprocessing import Process
import time
class MyProcess(Process):
    def __init__(self):
        super().__init__() # 初始化父类
    def run(self): # 重写父类方法，方法名必须为run
        for i  in range(5):
            time.sleep(2)
            print('...')
p = MyProcess()
p.start()
p.join() # 表示子进程结束再执行父进程的代码，括号中可以写出等待时间
print('父进程')

进程池（Pool）

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

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): #指定执行几次work
    p.apply_async(work)#非阻塞
    #p.apply(work)#阻塞
    print("添加成功")
p.close()#把池子关闭
p.join()

用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:
            num =q.get() #取出，取出一个删除一个
            print(num)
            if num == 9:
                break
p = Pool(3)
q = Manager().Queue()
p.apply_async(write,(q,))
p.apply_async(read,(q,))
p.close()
p.join()

线程(threading)

• 进程，能够完成多任务，比如 在一台电脑上能够同时运行多个QQ
• 线程，能够完成多任务，比如 一个QQ中的多个聊天窗口

import time
from threading import Thread
def say():
    time.sleep(1)
    print("我错了，我能站起来了吗？")

for i in range(5):
    #say()
    t = Thread(target=say) #开启一个线程
    t.start()

print("哈哈哈")

• 面向对象线程

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() 
t1 = MyThread()
t1.start()

多线程修改变量

多线程全局变量

from threading import Thread
import threading 
import time
num = 0
def test1():
    global num #修改全局变量需要提前声明
    num+=1
    time.sleep(1)
    print(threading.currentThread().name)
    print('test1',num)
def test2():
    time.sleep(3)
    print("test2",num)

t1 = Thread(target=test1)
t1.start()
t2 = Thread(target=test2)
t2.start()
# 多进程不共享全局变量，多线程共享全局变量

线程修改全局变量

from threading import Thread
import time
num = 0
def test1():
    global num
    for i in range(1000000):
        num+=1
    print(num)

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

t1 = Thread(target=test1)
t1.start()
#time.sleep(3)
t2 =Thread(target=test2)
t2.start()
# 不能使用多线程修改全局变量

多线程局部变量

import threading 
import time
def test1():
    num = 0
    if threading.currentThread().name == 'Thread-1':
        num+=1
        time.sleep(1)
        print('threading-1',num)
    else:
        time.sleep(3)
        print(num)
t1 = Thread(target = test1)
t1.start()
t2 = Thread(target = test1)
t2.start()

加锁

from threading import Thread
import 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(3)
t2 =Thread(target=test2)
t2.start()

互斥锁

• m.acquire() m.release() 调用一次释放，两个的位置决定是单线程和多线程

from threading import Thread,Lock
import time
num = 0
def test1():
    global num
    #m.acquire() 单线程
    for i in range(1000000):
        m.acquire() #多线程
        num+=1
        m.release()
    #m.release()
    print(num)

def test2():
    global num
    #m.acquire()
    for i in range(1000000):
        m.acquire()
        num+=1
        m.release()
    #m.release()
    print(num)

m = Lock()
t1 = Thread(target=test1)
t1.start()
#time.sleep(3)
t2 =Thread(target=test2)
t2.start()

死锁

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()

同步

• 同步就是协同步调，按预定的先后次序多个线程有序执行。

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()

生产者与消费者

from threading import Thread
from queue import Queue
import time


def producer():
    global q
    while True:
        if q.qsize() < 1000:
            for i in range(100):
                q.put("生成"+str(i))
                print("生成")
        time.sleep(0.5)


def consumer():
    global q
    while True:
        if q.qsize() > 600:
            for i in range(3):
                print("消费"+q.get())
        time.sleep(1)


q = Queue()
for i in range(500):
    q.put("初始化"+str(i))

for i in range(2):
    t1 = Thread(target=producer)
    t1.start()

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

异步

from multiprocessing import Pool
import time

def test():
    for i in range(5):
        time.sleep(0.5)
        print("做饭中")
    return "饭做好了"


def test2(msg):
    print(msg)

p = Pool()
p.apply_async(func = test,callback = test2)

for i in range(10):
    print("做作业")
time.sleep(0.5)