并发编程（四十六）

http://www.cnblogs.com/linhaifeng/articles/6817679.html

进程是对正在运行程序的一个抽象。即使可以利用的cpu只有一个（早期的计算机确实如此），也能保证支持（伪）并发的能力。

将一个单独的cpu变成多个虚拟的cpu（多道技术：时间多路复用和空间多路复用+硬件上支持隔离），没有进程的抽象，现代计算机将不复存在。

#一 操作系统的作用：
    1：隐藏丑陋复杂的硬件接口，提供良好的抽象接口
    2：管理、调度进程，并且将多个进程对硬件的竞争变得有序

#二 多道技术：
    1.产生背景：针对单核，实现并发
    ps：
    现在的主机一般是多核，那么每个核都会利用多道技术
    有4个cpu，运行于cpu1的某个程序遇到io阻塞，会等到io结束再重新调度，会被调度到4个
    cpu中的任意一个，具体由操作系统调度算法决定。
    
    2.空间上的复用：如内存中同时有多道程序
    3.时间上的复用：复用一个cpu的时间片
       强调：遇到io切，占用cpu时间过长也切，核心在于切之前将进程的状态保存下来，这样
            才能保证下次切换回来时，能基于上次切走的位置继续运行

线程

import threading
import time
def listen(parm):
    print("%s start listening %s... " %(time.ctime(),parm))
    time.sleep(3)
    print("%s end listening %s" %(time.ctime(),parm))

def play():
    print("%s start play...." %(time.ctime()))
    time.sleep(5)
    print("%s end play..." %(time.ctime()))


if __name__ == "__main__":
    t1 = threading.Thread(target=listen, args=("music",))
    t2 = threading.Thread(target=play)

    t1.start()
    t2.start()

    print("%s end of main..." %time.ctime())

'''
Fri Mar  8 17:14:31 2019 start listening music... 
Fri Mar  8 17:14:31 2019 start play....Fri Mar  8 17:14:31 2019 end of main...

Fri Mar  8 17:14:34 2019 end listening music
Fri Mar  8 17:14:36 2019 end play...
'''

join()方法

主线程要等待join的线程结束之后才能退出

import threading
import time
def listen(parm):
    print("%s start listening %s... " %(time.ctime(),parm))
    time.sleep(3)
    print("%s end listening %s" %(time.ctime(),parm))

def play():
    print("%s start play...." %(time.ctime()))
    time.sleep(5)
    print("%s end play..." %(time.ctime()))


if __name__ == "__main__":
    t1 = threading.Thread(target=listen, args=("music",))
    t2 = threading.Thread(target=play)

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    print("%s end of main..." %time.ctime())

'''
Fri Mar  8 17:18:01 2019 start listening music... 
Fri Mar  8 17:18:01 2019 start play....
Fri Mar  8 17:18:04 2019 end listening music
Fri Mar  8 17:18:06 2019 end play...
Fri Mar  8 17:18:06 2019 end of main...
'''

View Code

import threading
import time
def listen(parm):
    print("%s start listening %s... " %(time.ctime(),parm))
    time.sleep(3)
    print("%s end listening %s" %(time.ctime(),parm))

def play():
    print("%s start play...." %(time.ctime()))
    time.sleep(5)
    print("%s end play..." %(time.ctime()))


if __name__ == "__main__":
    t1 = threading.Thread(target=listen, args=("music",))
    t2 = threading.Thread(target=play)

    t1.start()
    t2.start()

    t1.join()
    #t2.join()

    print("%s end of main..." %time.ctime())

'''
Fri Mar  8 17:20:32 2019 start listening music... 
Fri Mar  8 17:20:32 2019 start play....
Fri Mar  8 17:20:35 2019 end listening music
Fri Mar  8 17:20:35 2019 end of main...
Fri Mar  8 17:20:37 2019 end play...
'''

View Code

守护线程Daemon()

必须在线程start()之前设置，守护线程与主线程同时结束

import threading
import time
def listen(parm):
    print("%s start listening %s... " %(time.ctime(),parm))
    time.sleep(3)
    print("%s end listening %s" %(time.ctime(),parm))

def play():
    print("%s start play...." %(time.ctime()))
    time.sleep(5)
    print("%s end play..." %(time.ctime()))


if __name__ == "__main__":
    t1 = threading.Thread(target=listen, args=("music",))
    t2 = threading.Thread(target=play)

    t2.setDaemon(True)

    t1.start()
    t2.start()

    print("%s end of main..." %time.ctime())

'''
Fri Mar  8 17:22:34 2019 start listening music... 
Fri Mar  8 17:22:34 2019 start play....Fri Mar  8 17:22:34 2019 end of main...

Fri Mar  8 17:22:37 2019 end listening music
'''

View Code

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁

需要强调的是：运行完毕并非终止运行

#1.对主进程来说，运行完毕指的是主进程代码运行完毕

#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

#1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，

#2 主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

迷惑人的例子
注意下面两个例子的时间

from threading import Thread
import time
def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


t1=Thread(target=foo)
t2=Thread(target=bar)

t1.daemon=True
t1.start()
t2.start()
print("main-------")
'''
123
456main-------

end123
end456
'''

View Code

from threading import Thread
import time
def foo():
    print(123)
    time.sleep(4)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


t1=Thread(target=foo)
t2=Thread(target=bar)

t1.daemon=True
t1.start()
t2.start()
print("main-------")
'''
123
456main-------

end456

Process finished with exit code 0
'''

View Code

threading.activeCount()，

getName()

getName()

import threading
import time
def listen(parm):
    print("%s start listening %s... " %(time.ctime(),parm))
    time.sleep(3)
    print("%s end listening %s" %(time.ctime(),parm))

def play():
    print("%s start play...." %(time.ctime()))
    time.sleep(5)
    print("%s end play..." %(time.ctime()))


if __name__ == "__main__":
    t1 = threading.Thread(target=listen, args=("music",))
    t2 = threading.Thread(target=play)

    t2.setDaemon(True)

    t1.start()
    t2.start()

    print(threading.activeCount()) # 3
    print(t1.getName()) # Thread-1
    print(t2.getName()) # Thread-1
    
    t1.setName("hello")
    print(t1.getName()) # hello

    print("%s end of main..." %time.ctime())

View Code

开启线程的两种方式

import threading
class MyThread(threading.Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name
    def run(self):
        print("start thread...")

t = MyThread('t1')
t.start()
print('main')
'''
start thread...
main
'''

View Code

并发与并行

并发：是指系统具有处理多个任务（动作）的能力

并行：是指系统具有同时处理多个任务（动作）的能力

并行是并发的一个子集

同步与异步

同步：当进程执行到一个IO（等待外部数据）的时候你，------等：同步

异步：.............................................................................，------不等：一直等到数据接收成功，再回来处理

GIL：全局解释锁

　　　因为GIL，所以，同一时刻，只有一个线程被CPU执行

任务可以分为：IO密集型

　　　　　　　计算密集型

对于IO密集型任务：Python的多线程是有意义的

　　　　　　　　　　也可采用多进程 + 协程

对于计算密集型的任务：Python的多线程就不推荐了，不适用了

对于计算密集型从下面例子可以看出来，开线程花费的时间与串行执行差不多

import threading
import time
def add():
    sum = 0
    for i in range(10000000):
        sum += i
    print("sum: %s" %sum)

def mul():
    m = 1
    for i in range(1,100000):
        m *= i
    print("mul: %s" %m)

t1 = threading.Thread(target=add)
t2 = threading.Thread(target=mul)

l = []
l.append(t1)
l.append(t2)

start = time.time()
for i in l:
    i.start()
for i in l:
    i.join()

end = time.time()

print("cost : %s" %(end - start))

'''
sum: 49999995000000
mul: 数字太大了。。。
cost : 7.897883892059326
'''

View Code

import threading
import time
def add():
    sum = 0
    for i in range(10000000):
        sum += i
    print("sum: %s" %sum)

def mul():
    m = 1
    for i in range(1,100000):
        m *= i
    print("mul: %s" %m)

t1 = threading.Thread(target=add)
t2 = threading.Thread(target=mul)

l = []
l.append(t1)
l.append(t2)

start = time.time()
# for i in l:
#     i.start()
# for i in l:
#     i.join()

add()
mul()

end = time.time()

print("cost : %s" %(end - start))

'''
sum: 49999995000000
mul: 数字太大了。。。
cost : 7.443098545074463
'''

View Code

同步锁

#1.线程抢的是GIL锁，GIL锁相当于执行权限，拿到执行权限后才能拿到互斥锁Lock，其他线程也可以抢到GIL，但如果发现Lock仍然没有被释放则阻塞，即便是拿到执行权限GIL也要立刻交出来

#2.join是等待所有，即整体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁都可以实现，毫无疑问，互斥锁的部分串行效率要更高

GIL VS Lock
    机智的同学可能会问到这个问题，就是既然你之前说过了，Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock? 
　首先我们需要达成共识：锁的目的是为了保护共享的数据，同一时间只能有一个线程来修改共享的数据
    然后，我们可以得出结论：保护不同的数据就应该加不同的锁。
　最后，问题就很明朗了，GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock
过程分析：所有线程抢的是GIL锁，或者说所有线程抢的是执行权限
　　线程1抢到GIL锁，拿到执行权限，开始执行，然后加了一把Lock，还没有执行完毕，即线程1还未释放Lock，有可能线程2抢到GIL锁，开始执行，执行过程中发现Lock还没有被线程1释放，于是线程2进入阻塞，被夺走执行权限，有可能线程1拿到GIL，然后正常执行到释放Lock。。。这就导致了串行运行的效果
　　既然是串行，那我们执行
　　t1.start()
　　t1.join
　　t2.start()
　　t2.join()
　　这也是串行执行啊，为何还要加Lock呢，需知join是等待t1所有的代码执行完，相当于锁住了t1的所有代码，而Lock只是锁住一部分操作共享数据的代码。

因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次

全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，

py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，

结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，

这样就解决了上述的问题，这可以说是Python早期版本的遗留问题。

import threading
import time
Num = 100
def sub():
    global Num
    temp = Num
    time.sleep(0.001)
    Num = temp - 1

l = []
for i in range(100):
    t = threading.Thread(target=sub)
    t.start()
    l.append(t)

for t in l:
    t.join()

print(Num) # 可能为93

锁通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，

调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，

再调用release方法释放锁：

import threading

R=threading.Lock()

R.acquire()
'''
对公共数据的操作
'''
R.release()

import threading
import time
lock = threading.Lock()
Num = 100
def sub():
    global Num
    lock.acquire()
    temp = Num
    time.sleep(0.001)
    Num = temp - 1
    lock.release()

l = []
for i in range(100):
    t = threading.Thread(target=sub)
    t.start()
    l.append(t)

for t in l:
    t.join()

print(Num) #结果肯定为0，由原来的并发执行变成串行，牺牲了执行效率保证了数据安全

View Code

分析：
　　#1.100个线程去抢GIL锁，即抢执行权限

#2. 肯定有一个线程先抢到GIL（暂且称为线程1），然后开始执行，一旦执行就会拿到lock.acquire()

#3. 极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发现互斥锁lock还未被线程1释放，

　　于是阻塞，被迫交出执行权限，即释放GIL

#4.直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程

互斥锁与join的区别（重点！！！）

#不加锁:并发执行,速度快,数据不安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    global n
    print('%s is running' %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()

    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:0.5216062068939209 n:99
'''


#不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全
from threading import current_thread,Thread,Lock
import os,time
def task():
    #未加锁的代码并发运行
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    #加锁的代码串行运行
    lock.acquire()
    temp=n
    time.sleep(0.5)
    n=temp-1
    lock.release()

if __name__ == '__main__':
    n=100
    lock=Lock()
    threads=[]
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 is running
Thread-2 is running
......
Thread-100 is running
主:53.294203758239746 n:0
'''

#有的同学可能有疑问:既然加锁会让运行变成串行,那么我在start之后立即使用join,就不用加锁了啊,也是串行的效果啊
#没错:在start之后立刻使用jion,肯定会将100个任务的执行变成串行,毫无疑问,最终n的结果也肯定是0,是安全的,但问题是
#start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的
#单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.
from threading import current_thread,Thread,Lock
import os,time
def task():
    time.sleep(3)
    print('%s start to run' %current_thread().getName())
    global n
    temp=n
    time.sleep(0.5)
    n=temp-1


if __name__ == '__main__':
    n=100
    lock=Lock()
    start_time=time.time()
    for i in range(100):
        t=Thread(target=task)
        t.start()
        t.join()
    stop_time=time.time()
    print('主:%s n:%s' %(stop_time-start_time,n))

'''
Thread-1 start to run
Thread-2 start to run
......
Thread-100 start to run
主:350.6937336921692 n:0 #耗时是多么的恐怖
'''

View Code

死锁现象与递归锁

死锁

所谓死锁：是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。

此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

import threading
import time

class MyThread(threading.Thread):

    def run(self):
        self.ActionA()
        self.ActionB()

    def ActionA(self):
        A.acquire()
        print(self.name, 'got A', time.ctime())
        time.sleep(2)

        B.acquire()
        print(self.name, 'got B', time.ctime())
        time.sleep(1)

        A.release()
        B.release()

    def ActionB(self):
        B.acquire()
        print(self.name, 'got B', time.ctime())
        time.sleep(2)

        A.acquire()
        print(self.name, 'got A', time.ctime())
        time.sleep(1)

        A.release()
        B.release()

if __name__ == "__main__":
    A = threading.Lock()
    B = threading.Lock()

    l = []
    for t in range(5):
        t = MyThread()
        l.append(t)
        t.start()
    for t in l:
        t.join()

    print('end of main...')
'''
Thread-1 got A Sat Mar  9 11:05:13 2019
Thread-1 got B Sat Mar  9 11:05:15 2019
Thread-1 got B Sat Mar  9 11:05:16 2019
Thread-2 got A Sat Mar  9 11:05:16 2019

分析：
此时线程1获得了B没有释放，希望再获得A，而线程2获得了A没有释放，希望再获得A
这时候就发生死锁了，程序不会再往下执行了
'''

View Code

递归锁

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。

直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

import threading
import time

class MyThread(threading.Thread):

    def run(self):
        self.ActionA()
        self.ActionB()

    def ActionA(self):
        R.acquire()
        print(self.name, 'got A', time.ctime())
        time.sleep(2)

        R.acquire()
        print(self.name, 'got B', time.ctime())
        time.sleep(1)

        R.release()
        R.release()

    def ActionB(self):
        R.acquire()
        print(self.name, 'got B', time.ctime())
        time.sleep(2)

        R.acquire()
        print(self.name, 'got A', time.ctime())
        time.sleep(1)

        R.release()
        R.release()

if __name__ == "__main__":
    R = threading.RLock()

    l = []
    for t in range(5):
        t = MyThread()
        l.append(t)
        t.start()
    for t in l:
        t.join()

    print('end of main...')
'''
Thread-1 got A Sat Mar  9 11:13:36 2019
Thread-1 got B Sat Mar  9 11:13:38 2019
Thread-2 got A Sat Mar  9 11:13:39 2019
Thread-2 got B Sat Mar  9 11:13:41 2019
Thread-2 got B Sat Mar  9 11:13:42 2019
Thread-2 got A Sat Mar  9 11:13:44 2019
Thread-4 got A Sat Mar  9 11:13:45 2019
Thread-4 got B Sat Mar  9 11:13:47 2019
Thread-4 got B Sat Mar  9 11:13:48 2019
Thread-4 got A Sat Mar  9 11:13:50 2019
Thread-1 got B Sat Mar  9 11:13:51 2019
Thread-1 got A Sat Mar  9 11:13:53 2019
Thread-3 got A Sat Mar  9 11:13:54 2019
Thread-3 got B Sat Mar  9 11:13:56 2019
Thread-5 got A Sat Mar  9 11:13:57 2019
Thread-5 got B Sat Mar  9 11:13:59 2019
Thread-5 got B Sat Mar  9 11:14:00 2019
Thread-5 got A Sat Mar  9 11:14:02 2019
Thread-3 got B Sat Mar  9 11:14:03 2019
Thread-3 got A Sat Mar  9 11:14:05 2019
end of main...
'''

View Code

mutexA=mutexB=threading.RLock() #一个线程拿到锁，counter加1,该线程内又碰到加锁的情况，则counter继续加1，这期间所有其他线程都只能等待，等待该线程释放所有锁，即counter递减到0为止

event同步对象

同进程的一样

线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其他线程需要通过判断某个线程的状态来确定自己下一步的操作,

这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。对象包含一个可由线程设置的信号标志,

它允许线程等待某些事件的发生。在初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,

那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。

如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行

event.isSet()：返回event的状态值；

event.wait()：如果 event.isSet()==False将阻塞线程；

event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态， 等待操作系统调度；

event.clear()：恢复event的状态值为False

import threading
import time
import random

def conn_mysql():
    count = 1
    while not event.is_set():
        if count > 3:
            raise TimeoutError
        print("[%s]第[%s]次开始尝试链接" %(threading.current_thread().getName(), count))
        event.wait(0.5)
        count += 1
    print("[%s]链接成功" % (threading.current_thread().getName()))

def check_mysql():
    print('33[45m[%s]正在检查mysql33[0m' %threading.current_thread().getName())
    time.sleep(random.randint(2,4))
    event.set()

if __name__ == "__main__":
    event = threading.Event()

    t1 = threading.Thread(target=conn_mysql)
    t2 = threading.Thread(target=conn_mysql)
    t3 = threading.Thread(target=check_mysql())

    t1.start()
    t2.start()
    t3.start()
'''
[MainThread]正在检查mysql
[Thread-1]链接成功
[Thread-2]链接成功
'''

View Code

信号量Semaphore

同进程的一样

Semaphore管理一个内置的计数器，
每当调用acquire()时内置计数器-1；
调用release() 时内置计数器+1；
计数器不能小于0；当计数器为0时，acquire()将阻塞线程直到其他线程调用release()。

实例：(同时只有3个线程可以获得semaphore,即可以限制最大连接数为3)：

import threading
import time
class MyThread(threading.Thread):
    def run(self):
        if semaphore.acquire():
            print("%s" %threading.current_thread().getName())
            time.sleep(2)
            semaphore.release()

if __name__ == "__main__":
    semaphore = threading.Semaphore(3) # 同时最多可以有三个线程进入临界区

    ths = []
    for t in range(15):
        t = MyThread()
        ths.append(t)
        t.start()

    for i in ths:
        i.join()
'''
一次三个线程
Thread-1
Thread-2
Thread-3
=======================
Thread-4
Thread-5
Thread-6
========================
...
'''

View Code

与进程池是完全不同的概念，进程池Pool(4)，最大只能产生4个进程，而且从头到尾都只是这四个进程，不会产生新的，而信号量是产生一堆线程/进程

互斥锁与信号量推荐博客

多线程利器——队列

队列有三种模式：FIFO，FILO，PriorityQueue（指定优先级）

FIFO:

import queue

q = queue.Queue(3) # 可以指定size FIFO

q.put(5)
q.put('hello')
q.put({"name": "zhangsan"})
# q.put(8) 不会引发异常，会一直阻塞在这里，直到有其他线程get其中一个数据之后
# q.put(6,False) 引发异常raise Full

while True:
    data = q.get() # # 队列数据取完之后会一直等待其他线程put数据
    #data = q.get(block=False) # 队列数据取完之后会引发异常 raise Empty
    print(data)
    print("----------")

'''
5
----------
hello
----------
{'name': 'zhangsan'}
----------
'''

View Code

FILO:

import queue

q = queue.LifoQueue() # 可以指定size FIFO

q.put(5)
q.put('hello')
q.put({"name": "zhangsan"})
# q.put(8) 不会引发异常，会一直阻塞在这里，直到有其他线程get其中一个数据之后
# q.put(6,False) 引发异常raise Full

while True:
    data = q.get() # # 队列数据取完之后会一直等待其他线程put数据
    #data = q.get(block=False) # 队列数据取完之后会引发异常 raise Empty
    print(data)
    print("----------")

'''
{'name': 'zhangsan'}
----------
hello
----------
5
----------
'''

View Code

PriorityQueue：

import queue

q = queue.PriorityQueue()

q.put([4,5])
q.put([6,'hello'])
q.put([2, {"name": "zhangsan"}])

while True:
    data = q.get()
    print(data)
    print("----------")

'''
[2, {'name': 'zhangsan'}]
----------
[4, 5]
----------
[6, 'hello']
----------
'''

View Code

生产者消费者模型

import threading
import time
import queue

q = queue.Queue()

def Coutomer(name):
    count = 0
    time.sleep(3)

    q.join()
    data = q.get()
    print("[%s] is coutom %s号cake" %(name, data))




def Producer(name):
    count = 0
    while count < 10:
        print("[%s] is making cake..." %name)
        time.sleep(1)
        q.put(count)
        q.task_done()
        print("produce %s" %count)
        count += 1
    print("ok")


if __name__ == "__main__":

    t1 = threading.Thread(target=Producer, args=("Cooker",))
    t2 = threading.Thread(target=Coutomer, args=("C1君",))
    t3 = threading.Thread(target=Coutomer, args=("C2君",))
    t4 = threading.Thread(target=Coutomer, args=("C3君",))
    t5 = threading.Thread(target=Coutomer, args=("C4君",))

    thr = [t1, t2, t3, t4, t5]

    for t in thr:
        t.start()
    for t in thr:
        t.join()

'''
[Cooker] is making cake...
produce 0
[Cooker] is making cake...
produce 1
[Cooker] is making cake...
[C2君] is coutom 0号cake
[C1君] is coutom 1号cake
produce 2
[Cooker] is making cake...
[C4君] is coutom 2号cake
produce 3
[Cooker] is making cake...
[C3君] is coutom 3号cake
produce 4
[Cooker] is making cake...
produce 5
[Cooker] is making cake...
produce 6
[Cooker] is making cake...
produce 7
[Cooker] is making cake...
produce 8
[Cooker] is making cake...
produce 9
ok
'''

View Code

多进程的调用

from multiprocessing import Process
import time

def func(name):
    print("[%s] is working" %name, time.ctime())
    time.sleep(1)

if __name__ == "__main__":
    p1 = Process(target=func, args=("A",))
    p2 = Process(target=func, args=("B",))
    p3 = Process(target=func, args=("C",))

    p = [p1, p2, p3]
    for i in p:
        i.start()
    for i in p:
        i.join()
    print("ending...")
'''
[C] is working Sat Mar  9 17:40:50 2019
[B] is working Sat Mar  9 17:40:50 2019
[A] is working Sat Mar  9 17:40:50 2019
ending...
'''

View Code

守护进程

from multiprocessing import Process
import time

def func(name):
    print("[%s] is working" %name, time.ctime())
    time.sleep(1)
    print("[%s] is ending" %name, time.ctime())

if __name__ == "__main__":
    p1 = Process(target=func, args=("A",))
    p2 = Process(target=func, args=("B",))
    p3 = Process(target=func, args=("C",))

    p = [p1, p2, p3]
    p1.daemon = True
    for i in p:
        i.start()
    # for i in p:
    #     i.join()
    print("ending...")
'''
分析：设置进程A为守护进程，与主进程一起结束
ending...
[B] is working Sat Mar  9 17:44:53 2019
[C] is working Sat Mar  9 17:44:53 2019
[B] is ending Sat Mar  9 17:44:54 2019
[C] is ending Sat Mar  9 17:44:54 2019
ending...
'''

View Code

from multiprocessing import Process
import time
import os

def info(title):
    print("title: %s" %title)
    print("parent id: %s" %os.getppid())
    print("my id: %s" %os.getpid())

def func(title):
    info(title)
    print("hello")

if __name__ == "__main__":
    func("zhangsan")
    print("-"*10)
    p = Process(target=func, args=("lisi",))
    p.start()
    p.join()
'''
title: zhangsan
parent id: 14372
my id: 15148
hello
----------
title: lisi
parent id: 15148
my id: 5280
hello
'''

View Code