python学习笔记09--线程、进程

 本节内容

　　一、进程与线程的概念

　　　　1.1进程

　　　　1.2线程

　　　　1.3进程与线程的区别

　　二、线程

　　　　2、1启一个线程

　　　　2.2线程的2种调用方式

　　　　2.3 join

　　　　2.4 守护线程Daemon

　　　　2.5线程锁

　　　　2.6全局解析锁

　　　　2.7递归锁

　　　　2.8Semaphore(信号量)

　　　　2.9事件Events（红绿灯举例）

　　　　2.10queue队列

　　　　2.11生产消费者模型

　　三、进程

　　　　3.1多进程

　　　　3.2进程间通讯

　　　　3.3进程池

一、进程与线程的概念

1.1进程

什么是进程(process)？

　　以QQ为例， QQ 要以一个整体的形式暴露给操作系统管理，里面包含对各种资源的调用，内存的管理，网络接口的调用等。。。。对各种资源管理的集合 就可以称为： 进程。

　　程序的执行实例称为进程。每个进程都提供执行程序所需的资源。 进程具有虚拟地址空间，可执行代码，系统对象的打开句柄，安全上下文，唯一进程标识符，环境变量，优先级类，最小和最大工作集大小以及至少一个执行线程。 每个进程都使用单个线程启动，通常称为主线程，但可以从其任何线程创建其他线程。
　　程序并不能单独运行，只有将程序装载到内存中，系统为它分配资源才能运行，而这种执行的程序就称之为进程。程序和进程的区别就在于：程序是指令的集合，它是进程运行的静态描述文本；进程是程序的一次执行活动，属于动态概念。在多道编程中，我们允许多个程序同时加载到内存中，在操作系统的调度下，可以实现并发地执行。这是这样的设计，大大提高了CPU的利用率。进程的出现让每个用户感觉到自己独享CPU，因此，进程就是为了在CPU上实现多道编程而提出的。

有了进程为什么还要线程？

　　进程有很多优点，它提供了多道编程，让我们感觉我们每个人都拥有自己的CPU和其他资源，可以提高计算机的利用率。很多人就不理解了，既然进程这么优秀，为什么还要线程呢？其实，仔细观察就会发现进程还是有很多缺陷的，主要体现在两点上：

• 进程只能在一个时间干一件事，如果想同时干两件事或多件事，进程就无能为力了。

• 进程在执行的过程中如果阻塞，例如等待输入，整个进程就会挂起，即使进程中有些工作不依赖于输入的数据，也将无法执行。

例如，我们在使用qq聊天， qq做为一个独立进程如果同一时间只能干一件事，那他如何实现在同一时刻 即能监听键盘输入、又能监听其它人给你发的消息、同时还能把别人发的消息显示在屏幕上呢？你会说，操作系统不是有分时么？但我的亲，分时是指在不同进程间的分时呀， 即操作系统处理一会你的qq任务，又切换到word文档任务上了，每个cpu时间片分给你的qq程序时，你的qq还是只能同时干一件事呀。

再直白一点， 一个操作系统就像是一个工厂，工厂里面有很多个生产车间，不同的车间生产不同的产品，每个车间就相当于一个进程，且你的工厂又穷，供电不足，同一时间只能给一个车间供电，为了能让所有车间都能同时生产，你的工厂的电工只能给不同的车间分时供电，但是轮到你的qq车间时，发现只有一个干活的工人，结果生产效率极低，为了解决这个问题，应该怎么办呢？。。。。没错，你肯定想到了，就是多加几个工人，让几个人工人并行工作，这每个工人，就是线程！

 1.2线程

线程: 是操作系统最小的调度单位， 是一串指令的集合。是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。官网链接：https://docs.python.org/3/library/threading.html?highlight=threading#

线程是执行上下文，它是CPU执行指令流所需的所有信息。
假设你正在读一本书，而你现在想休息一下，但是你希望能够从你停下来的确切位置回来并继续阅读。实现这一目标的一种方法是记下页码，行号和字号。因此，阅读书籍的执行环境就是这三个数字。如果你有一个室友，并且她使用相同的技术，她可以在你不使用时拿走这本书，并从她停下的地方继续阅读。然后你可以把它拿回来，并从你原来的地方恢复。
线程以相同的方式工作。 CPU正在给你一种错觉，即它同时进行多次计算。它通过在每次计算上花费一点时间来做到这一点。它可以这样做，因为它具有每个计算的执行上下文。就像您可以与朋友共享一本书一样，许多任务可以共享CPU。在更技术层面上，执行上下文（因此是一个线程）由CPU寄存器的值组成。最后：线程与进程不同。线程是执行的上下文，而进程是与计算相关联的一堆资源。一个进程可以有一个或多个线程。澄清：与进程相关联的资源包括内存页面（进程中的所有线程具有相同的内存视图），文件描述符（例如，打开套接字）和安全凭证（例如，启动该进程的用户的ID）处理）。

 1.3进程与线程的区别：

进程 要操作cpu , 必须要先创建一个线程 ，所有在同一个进程里的线程是共享同一块内存空间的。

1、线程共享创建它的进程的地址空间; 进程有自己的地址空间。（线程共享内存空间，进程的内存是独立的）
2、线程可以直接访问其进程的数据段; 进程拥有自己父进程数据段的副本。
3、线程可以直接与其进程的其他线程通信; 进程必须使用进程间通信来与兄弟进程通信。（同一个进程的线程之间可以直接交流，两个进程想通信，必须通过一个中间代理来实现）
4、创建新线程很简单， 创建新进程需要对其父进程进行一次克隆。
5、一个线程可以控制和操作同一进程里的其他线程，但是进程只能操作子进程。
6、对主线程的更改（取消，优先级更改等）可能会影响进程的其他线程的行为; 对父进程的更改不会影响子进程。

二、线程

2.1启动一个线程

举例子，演示一个最简单的多线程

import threading
import time

def run(n):
    print("task",n)
    time.sleep(2)
    
t1 = threading.Thread(target=run,args=("t1",))
t2 = threading.Thread(target=run,args=("t2",))
#总共运行了2秒，因为是并行的2秒
t1.start()
t2.start()

#比较,这里会运行4秒，是先后各两秒
# run(t1)
# run(t2)

 2、2线程的2种调用方式

Python threading模块线程有2种调用方式，如下：

import threading
import time


def sayhi(num):  # 定义每个线程要运行的函数

    print("running on number:%s" % num)

    time.sleep(3)


if __name__ == '__main__':
    t1 = threading.Thread(target=sayhi, args=(1,))  # 生成一个线程实例
    t2 = threading.Thread(target=sayhi, args=(2,))  # 生成另一个线程实例

    t1.start()  # 启动线程
    t2.start()  # 启动另一个线程

    print(t1.getName())  # 获取线程名
    print(t2.getName())

import threading
import time


class MyThread(threading.Thread):
    def __init__(self, num):
        threading.Thread.__init__(self)
        self.num = num

    def run(self):  # 定义每个线程要运行的函数

        print("running on number:%s" % self.num)

        time.sleep(3)


if __name__ == '__main__':
    t1 = MyThread(1)
    t2 = MyThread(2)
    t1.start()
    t2.start()

 

2.3 join

import threading
import time

class MyThread(threading.Thread):
    def __init__(self,n,sleep_time):
        super(MyThread,self).__init__()
        self.n =  n
        self.sleep_time = sleep_time
    def run(self):
        print("runnint task ",self.n )
        time.sleep(self.sleep_time)
        print("task done,",self.n )


t1 = MyThread("t1",2)
t2 = MyThread("t2",4)

t1.start()
t2.start()

t1.join() #=wait()
t2.join()

print("main thread....")

import threading
import time

def run(n):
    print("task ",n )
    time.sleep(2)
    print("task done",n)

start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.start()
    t_objs.append(t) #为了不阻塞后面线程的启动，不在这里join，先放到一个列表里

for t in t_objs: #循环线程实例列表，等待所有线程执行完毕
    t.join()


print("----------all threads has finished...")
print("cost:",time.time() - start_time)
# run("t1")
# run("t2")

 在一个进程下开启多个线程与在一个进程下开启多个子进程的区别

from threading import Thread
from multiprocessing import Process
import os

def work():
    print('hello')

if __name__ == '__main__':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()
    print('主线程/主进程')
    '''
    打印结果:
    hello
    主线程/主进程
    '''

    #在主进程下开启子进程
    t=Process(target=work)
    t.start()
    print('主线程/主进程')
    '''
    打印结果:
    主线程/主进程
    hello
    '''

from threading import Thread
from multiprocessing import Process
import os

def work():
    print('hello',os.getpid())

if __name__ == '__main__':
    #part1:在主进程下开启多个线程,每个线程都跟主进程的pid一样
    t1=Thread(target=work)
    t2=Thread(target=work)
    t1.start()
    t2.start()
    print('主线程/主进程pid',os.getpid())

    #part2:开多个进程,每个进程都有不同的pid
    p1=Process(target=work)
    p2=Process(target=work)
    p1.start()
    p2.start()
    print('主线程/主进程pid',os.getpid())

from  threading import Thread
from multiprocessing import Process
import os
def work():
    global n
    n=0

if __name__ == '__main__':
    # n=100
    # p=Process(target=work)
    # p.start()
    # p.join()
    # print('主',n) #毫无疑问子进程p已经将自己的全局的n改成了0,但改的仅仅是它自己的,查看父进程的n仍然为100


    n=1
    t=Thread(target=work)
    t.start()
    t.join()
    print('主',n) #查看结果为0,因为同一进程内的线程之间共享进程内的数据

 练习

练习一：

#_*_coding:utf-8_*_
#!/usr/bin/env python
import multiprocessing
import threading

import socket
s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.bind(('127.0.0.1',8080))
s.listen(5)

def action(conn):
    while True:
        data=conn.recv(1024)
        print(data)
        conn.send(data.upper())

if __name__ == '__main__':

    while True:
        conn,addr=s.accept()


        p=threading.Thread(target=action,args=(conn,))
        p.start()

#_*_coding:utf-8_*_
#!/usr/bin/env python


import socket

s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect(('127.0.0.1',8080))

while True:
    msg=input('>>: ').strip()
    if not msg:continue

    s.send(msg.encode('utf-8'))
    data=s.recv(1024)
    print(data)

练习二：三个任务，一个接收用户输入，一个将用户输入的内容格式化成大写，一个将格式化后的结果存入文件

from threading import Thread
msg_l=[]
format_l=[]
def talk():
    while True:
        msg=input('>>: ').strip()
        if not msg:continue
        msg_l.append(msg)

def format_msg():
    while True:
        if msg_l:
            res=msg_l.pop()
            format_l.append(res.upper())

def save():
    while True:
        if format_l:
            with open('db.txt','a',encoding='utf-8') as f:
                res=format_l.pop()
                f.write('%s
' %res)

if __name__ == '__main__':
    t1=Thread(target=talk)
    t2=Thread(target=format_msg)
    t3=Thread(target=save)
    t1.start()
    t2.start()
    t3.start()

线程相关的其他方法

Thread实例对象的方法
  # isAlive(): 返回线程是否活动的。
  # getName(): 返回线程名。
  # setName(): 设置线程名。

threading模块提供的一些方法：
  # threading.currentThread(): 返回当前的线程变量。
  # threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
  # threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

from threading import Thread
import threading
from multiprocessing import Process
import os

def work():
    import time
    time.sleep(3)
    print(threading.current_thread().getName())


if __name__ == '__main__':
    #在主进程下开启线程
    t=Thread(target=work)
    t.start()

    print(threading.current_thread().getName())
    print(threading.current_thread()) #主线程
    print(threading.enumerate()) #连同主线程在内有两个运行的线程
    print(threading.active_count())
    print('主线程/主进程')

    '''
    打印结果:
    MainThread
    <_MainThread(MainThread, started 140735268892672)>
    [<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
    主线程/主进程
    Thread-1
    '''

主线程等待子线程结束

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.start()
    t.join()
    print('主线程')
    print(t.is_alive())
    '''
    egon say hello
    主线程
    False
    '''

2.4 守护线程Daemon

有些线程执行后台任务，例如发送keepalive数据包，或执行定期垃圾收集等等。 这些仅在主程序运行时才有用，并且一旦其他非守护程序线程退出就可以将它们终止。
如果没有守护程序线程，您必须跟踪它们，并在程序完全退出之前告诉它们退出。 通过将它们设置为守护程序线程，您可以让它们运行并忘记它们，当程序退出时，任何守护程序线程都会自动终止。

import threading
import time

def run(n):
    print("task ",n )
    time.sleep(2)
    print("task done",n,threading.current_thread())

start_time = time.time()
t_objs = [] #存线程实例
for i in range(50):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.setDaemon(True) #把当前线程设置为守护线程
    t.start()
    t_objs.append(t) #为了不阻塞后面线程的启动，不在这里join，先放到一个列表里

# for t in t_objs: #循环线程实例列表，等待所有线程执行完毕
#     t.join()

time.sleep(2)
print("----------all threads has finished...",threading.current_thread(),threading.active_count())
print("cost:",time.time() - start_time)
# run("t1")
# run("t2")

import time
import threading


def run(n):
    print('[%s]------running----
' % n)
    time.sleep(2)
    print('--done--')


def main():
    for i in range(5):
        t = threading.Thread(target=run, args=[i, ])
        t.start()
        t.join(1)
        print('starting thread', t.getName())


m = threading.Thread(target=main, args=[])
m.setDaemon(True)  # 将main线程设置为Daemon线程,它做为程序主线程的守护线程,当主线程退出时,m线程也会退出,由m启动的其它子线程会同时退出,不管是否执行完任务
m.start()
m.join(timeout=2)
print("---main thread done----")

注意：守护程序线程在关闭时突然停止。 他们的资源（例如打开文件，数据库事务等）可能无法正确发布。 如果您希望线程正常停止，请将它们设置为非守护进程并使用合适的信号机制（如Event）。

无论是进程还是线程，都遵循：守护xxx会等待主xxx运行完毕后被销毁
需要强调的是：运行完毕并非终止运行
#1.对主进程来说，运行完毕指的是主进程代码运行完毕
#2.对主线程来说，运行完毕指的是主线程所在的进程内所有非守护线程统统运行完毕，主线程才算运行完毕

详细解释：

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

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

from threading import Thread
import time
def sayhi(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=sayhi,args=('egon',))
    t.setDaemon(True) #必须在t.start()之前设置
    t.start()

    print('主线程')
    print(t.is_alive())
    '''
    主线程
    True
    '''

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

2.5线程锁(互斥锁Mutex)

 一个进程下可以启动多个线程，多个线程共享父进程的内存空间，也就意味着每个线程可以访问同一份数据，此时，如果2个线程同时要修改同一份数据，会出现什么状况？

import time
import threading


def addNum():
    global num  # 在每个线程中都获取这个全局变量
    print('--get num:', num)
    time.sleep(1)
    num -= 1  # 对此公共变量进行-1操作


num = 100  # 设定一个共享变量
thread_list = []
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)

for t in thread_list:  # 等待所有线程执行完毕
    t.join()

print('final num:', num)

正常来讲，这个num结果应该是0， 但在python 2.7上多运行几次，会发现，最后打印出来的num结果不总是0，为什么每次运行的结果不一样呢？ 哈，很简单，假设你有A,B两个线程，此时都 要对num 进行减1操作， 由于2个线程是并发同时运行的，所以2个线程很有可能同时拿走了num=100这个初始变量交给cpu去运算，当A线程去处完的结果是99，但此时B线程运算完的结果也是99，两个线程同时CPU运算的结果再赋值给num变量后，结果就都是99。那怎么办呢？ 很简单，每个线程在要修改公共数据时，为了避免自己在还没改完的时候别人也来修改此数据，可以给这个数据加一把锁， 这样其它线程想修改此数据时就必须等待你修改完毕并把锁释放掉后才能再访问此数据。 

*注：不要在3.x上运行，不知为什么，3.x上的结果总是正确的，可能是自动加了锁

import time
import threading


def addNum():
    global num  # 在每个线程中都获取这个全局变量
    print('--get num:', num)
    time.sleep(1)
    lock.acquire()  # 修改数据前加锁
    num -= 1  # 对此公共变量进行-1操作
    lock.release()  # 修改后释放


num = 100  # 设定一个共享变量
thread_list = []
lock = threading.Lock()  # 生成全局锁
for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)

for t in thread_list:  # 等待所有线程执行完毕
    t.join()

print('final num:', num)

import threading
import time

def run(n):
    lock.acquire()
    global  num
    num +=1
    time.sleep(1)
    lock.release()


lock = threading.Lock()
num = 0
t_objs = [] #存线程实例
for i in range(10):
    t = threading.Thread(target=run,args=("t-%s" %i ,))
    t.start()
    t_objs.append(t) #为了不阻塞后面线程的启动，不在这里join，先放到一个列表里

for t in t_objs: #循环线程实例列表，等待所有线程执行完毕
    t.join()

print("----------all threads has finished...",threading.current_thread(),threading.active_count())

print("num:",num)

 2.6全局解析锁 GIL VS Lock

 机智的同学可能会问到这个问题，就是既然你之前说过了，Python已经有一个GIL来保证同一时间只能有一个线程来执行了，为什么这里还需要lock? 注意啦，这里的lock是用户级的lock,跟那个GIL没关系 ，具体我们通过下图来看一下+配合我现场讲给大家，就明白了。

 

 那你又问了， 既然用户程序已经自己有锁了，那为什么C python还需要GIL呢？加入GIL主要的原因是为了降低程序的开发的复杂度，比如现在的你写python不需要关心内存回收的问题，因为Python解释器帮你自动定期进行内存回收，你可以理解为python解释器里有一个独立的线程，每过一段时间它起wake up做一次全局轮询看看哪些内存数据是可以被清空的，此时你自己的程序 里的线程和 py解释器自己的线程是并发运行的，假设你的线程删除了一个变量，py解释器的垃圾回收线程在清空这个变量的过程中的clearing时刻，可能一个其它线程正好又重新给这个还没来及得清空的内存空间赋值了，结果就有可能新赋值的数据被删除了，为了解决类似的问题，python解释器简单粗暴的加了锁，即当一个线程运行时，其它人都不能动，这样就解决了上述的问题，  这可以说是Python早期版本的遗留问题。

2.7递归锁 RLOCK

说白了就是在一个大锁中还要再包含子锁

__author__ = "Alex Li"

import threading, time


def run1():
    print("grab the first part data")
    lock.acquire()
    global num
    num += 1
    lock.release()
    return num


def run2():
    print("grab the second part data")
    lock.acquire()
    global num2
    num2 += 1
    lock.release()
    return num2


def run3():
    lock.acquire()
    res = run1()
    print('--------between run1 and run2-----')
    res2 = run2()
    lock.release()
    print(res, res2)




num, num2 = 0, 0
lock = threading.RLock()
for i in range(1):
    t = threading.Thread(target=run3)
    t.start()

while threading.active_count() != 1:
    print(threading.active_count())
else:
    print('----all threads done---')
    print(num, num2)

2.8Semaphore(信号量)

 互斥锁 同时只允许一个线程更改数据，而Semaphore是同时允许一定数量的线程更改数据 ，比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

__author__ = "Alex Li"

import threading, time


def run(n):
    semaphore.acquire()
    time.sleep(1)
    print("run the thread: %s
" % n)
    semaphore.release()


if __name__ == '__main__':
    semaphore = threading.BoundedSemaphore(5)  # 最多允许5个线程同时运行
    for i in range(22):
        t = threading.Thread(target=run, args=(i,))
        t.start()
while threading.active_count() != 1:
    pass  # print threading.active_count()
else:
    print('----all threads done---')
 

此类表示仅在经过一定时间后才应运行的操作 。与线程一样，通过调用start（）方法启动计时器。
 通过调用thecancel（）方法可以停止计时器（在其动作开始之前）。
计时器在执行其操作之前将等待的时间间隔可能与用户指定的时间间隔不完全相同。

2.9事件Events

事件是一个简单的同步对象;
该事件代表一个内部标志和线程
可以等待设置标志，或者自己设置或清除标志。
event = threading.Event（）
＃客户端线程可以等待设置标志
event.wait（）
#a服务器线程可以设置或重置它
event.set（）
event.clear（）
如果设置了标志，则wait方法不会执行任何操作。
如果该标志被清除，则等待将被阻塞，直到它再次被设置为止。
任意数量的线程都可以等待同一事件。

"""
同进程的一样

线程的一个关键特性是每个线程都是独立运行且状态不可预测。
如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。
为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,
它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象,
而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,
它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
"""

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

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

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

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

通过Event来实现两个或多个线程间的交互，下面是一个红绿灯的例子，即起动一个线程做交通指挥灯，生成几个线程做车辆，车辆行驶按红灯停，绿灯行的规则。

"""
例如，有多个工作线程尝试链接MySQL，我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器，
如果连接不成功，都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作


"""

from threading import Thread,Event
import threading
import time,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=Event()
    conn1=Thread(target=conn_mysql)
    conn2=Thread(target=conn_mysql)
    check=Thread(target=check_mysql)

    conn1.start()
    conn2.start()
    check.start()

__author__ = "Alex Li"

import time
import threading


event = threading.Event()

def lighter():
    count = 0
    event.set() #先设置绿灯
    while True:
        if count >5 and count < 10: #改成红灯
            event.clear() #把标志位清了
            print("33[41;1mred light is on....33[0m")
        elif count >10:
            event.set() #变绿灯
            count = 0
        else:
            print("33[42;1mgreen light is on....33[0m")
        time.sleep(1)
        count +=1

def car(name):
    while True:
        if event.is_set(): #代表绿灯
            print("[%s] running..."% name )
            time.sleep(1)
        else:
            print("[%s] sees red light , waiting...." %name)
            event.wait()
            print("33[34;1m[%s] green light is on, start going...33[0m" %name)


light = threading.Thread(target=lighter,)
light.start()

car1 = threading.Thread(target=car,args=("Tesla",))
car1.start()

这里还有一个event使用的例子，员工进公司门要刷卡， 我们这里设置一个线程是“门”， 再设置几个线程为“员工”，员工看到门没打开，就刷卡，刷完卡，门开了，员工就可以通过

#_*_coding:utf-8_*_
__author__ = 'Alex Li'
import threading
import time
import random

def door():
    door_open_time_counter = 0
    while True:
        if door_swiping_event.is_set():
            print("33[32;1mdoor opening....33[0m")
            door_open_time_counter +=1

        else:
            print("33[31;1mdoor closed...., swipe to open.33[0m")
            door_open_time_counter = 0 #清空计时器
            door_swiping_event.wait()


        if door_open_time_counter > 3:#门开了已经3s了,该关了
            door_swiping_event.clear()

        time.sleep(0.5)


def staff(n):

    print("staff [%s] is comming..." % n )
    while True:
        if door_swiping_event.is_set():
            print("33[34;1mdoor is opened, passing.....33[0m")
            break
        else:
            print("staff [%s] sees door got closed, swipping the card....." % n)
            print(door_swiping_event.set())
            door_swiping_event.set()
            print("after set ",door_swiping_event.set())
        time.sleep(0.5)
door_swiping_event  = threading.Event() #设置事件


door_thread = threading.Thread(target=door)
door_thread.start()



for i in range(5):
    p = threading.Thread(target=staff,args=(i,))
    time.sleep(random.randrange(3))
    p.start()

2.10queue队列

当必须在多个线程之间安全地交换信息时，队列在线程编程中特别有用。
class queue.Queue（maxsize = 0）＃先入先出
class queue.LifoQueue（maxsize = 0）#last in fisrt out  #后进先出
class queue.PriorityQueue（maxsize = 0）＃存储数据时可设置优先级的队列  #设置优先级
    优先级队列的构造函数。 maxsize是一个整数，用于设置可以放入队列的项目数的上限。达到此大小后，插入将阻止，直到消耗队列项。如果maxsize小于或等于零，则队列大小为无限大。
    首先检索最低值的条目（最低值条目是由sorted（list（entries））[0]返回的条目。条目的典型模式是以下形式的元组：（priority_number，data）。

exception queue.Empty       在对空的Queue对象调用非阻塞get（）（或get_nowait（））时引发异常。异常队列。

exception queue.Full   在已满的Queue对象上调用非阻塞put（）（或put_nowait（））时引发异常。

Queue.qsize（）

Queue.empty（）  #return如果为空则为真

Queue.full（）＃如果已满，则返回True

Queue.put（item，block = True，timeout = None）
    将项目放入队列。如果可选的args块为true且timeout为None（默认值），则在必要时阻塞，直到有空闲插槽可用。如果timeout是一个正数，则它会阻止最多超时秒，如果在该时间内没有可用的空闲槽，则会引发Full异常。否则（块为假），如果空闲插槽立即可用，则将项目放在队列中，否则引发完全异常（在这种情况下忽略超时）。

Queue.put_nowait（项目）     相当于put（item，False）。

Queue.get（block = True，timeout = None）
    从队列中删除并返回一个项目。如果可选的args块为true且timeout为None（默认值），则在必要时阻止，直到某个项可用为止。如果timeout是一个正数，则它会阻止最多超时秒，如果在该时间内没有可用的项，则会引发Empty异常。否则（块为假），如果一个项立即可用，则返回一个项，否则引发Empty异常（在这种情况下忽略超时）。

Queue.get_nowait（）    相当于get（False）。
提供了两种方法来支持跟踪守护进程消费者线程是否已完全处理入队任务。

Queue.task_done（）
    表明以前排队的任务已完成。由队列使用者线程使用。对于用于获取任务的每个get（），对task_done（）的后续调用会告知队列任务的处理已完成。
    如果join（）当前正在阻塞，则它将在所有项目都已处理后恢复（这意味着已为每个已放入队列的项目收到task_done（）调用）。
    如果调用的次数超过队列中放置的项目，则引发ValueError。

Queue.join（）阻止直到队列被消费完毕

看代码实现

import queue


q  = queue.Queue()

q.put(1)
q.put(2)
q.put(3)
print(q.get())
print(q.get())
print(q.get())

import queue


q  = queue.LifoQueue()

q.put(1)
q.put(2)
q.put(3)
print(q.get())
print(q.get())
print(q.get())

import queue

q = queue.PriorityQueue()

q.put((-1,"chenronghua"))
q.put((3,"hanyang"))
q.put((10,"alex"))
q.put((6,"wangsen"))

print(q.get())
print(q.get())
print(q.get())
print(q.get())

2.11生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里，生产者就是生产数据的线程，消费者就是消费数据的线程。在多线程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

__author__ = "Alex Li"
import threading,time

import queue

q = queue.Queue(maxsize=10)

def Producer(name):
    count = 1
    while True:
        q.put("骨头%s" % count)
        print("生产了骨头",count)
        count +=1
        time.sleep(0.1)



def  Consumer(name):
    #while q.qsize()>0:
    while True:
        print("[%s] 取到[%s] 并且吃了它..." %(name, q.get()))
        time.sleep(1)



p = threading.Thread(target=Producer,args=("Alex",))
c = threading.Thread(target=Consumer,args=("ChengRonghua",))
c1 = threading.Thread(target=Consumer,args=("王森",))


p.start()
c.start()
c1.start()

import threading
import queue


def producer():
    for i in range(10):
        q.put("骨头 %s" % i)

    print("开始等待所有的骨头被取走...")
    q.join()
    print("所有的骨头被取完了...")


def consumer(n):
    while q.qsize() > 0:
        print("%s 取到" % n, q.get())
        q.task_done()  # 告知这个任务执行完了


q = queue.Queue()

p = threading.Thread(target=producer, )
p.start()

c1 = consumer("李闯")

import time,random
import queue,threading
q = queue.Queue()
def Producer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(3))
    q.put(count)
    print('Producer %s has produced %s baozi..' %(name, count))
    count +=1
def Consumer(name):
  count = 0
  while count <20:
    time.sleep(random.randrange(4))
    if not q.empty():
        data = q.get()
        print(data)
        print('33[32;1mConsumer %s has eat %s baozi...33[0m' %(name, data))
    else:
        print("-----no baozi anymore----")
    count +=1
p1 = threading.Thread(target=Producer, args=('A',))
c1 = threading.Thread(target=Consumer, args=('B',))
p1.start()
c1.start()

了解

条件Condition

使得线程等待，只有满足某条件时，才释放n个线程

import threading
 
def run(n):
    con.acquire()
    con.wait()
    print("run the thread: %s" %n)
    con.release()
 
if __name__ == '__main__':
 
    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()
 
    while True:
        inp = input('>>>')
        if inp == 'q':
            break
        con.acquire()
        con.notify(int(inp))
        con.release()

def condition_func():

    ret = False
    inp = input('>>>')
    if inp == '1':
        ret = True

    return ret


def run(n):
    con.acquire()
    con.wait_for(condition_func)
    print("run the thread: %s" %n)
    con.release()

if __name__ == '__main__':

    con = threading.Condition()
    for i in range(10):
        t = threading.Thread(target=run, args=(i,))
        t.start()

定时器

定时器，指定n秒后执行某操作

from threading import Timer
 
 
def hello():
    print("hello, world")
 
t = Timer(1, hello)
t.start()  # after 1 seconds, "hello, world" will be printed

from threading import Timer
import random,time

class Code:
    def __init__(self):
        self.make_cache()

    def make_cache(self,interval=5):
        self.cache=self.make_code()
        print(self.cache)
        self.t=Timer(interval,self.make_cache)
        self.t.start()

    def make_code(self,n=4):
        res=''
        for i in range(n):
            s1=str(random.randint(0,9))
            s2=chr(random.randint(65,90))
            res+=random.choice([s1,s2])
        return res

    def check(self):
        while True:
            inp=input('>>: ').strip()
            if inp.upper() ==  self.cache:
                print('验证成功',end='
')
                self.t.cancel()
                break


if __name__ == '__main__':
    obj=Code()
    obj.check()

三、进程

3.1多进程multiprocessing
multiprocessing是一个使用类似于线程模块的API支持产生进程的包。
多处理包提供本地和远程并发，通过使用子进程而不是线程有效地侧向执行全局解释器锁。
 因此，多处理模块允许程序员充分利用给定机器上的多个处理器。 它可以在Unix和Windows上运行

example：先起一个进程，并且里面包含一个字进程

#起一个进程，方式与线程类似
from multiprocessing import Process
import time
import threading

def thread_run():
    print(threading.get_ident()) #获取进程ID
def func(name):
    time.sleep(2)
    print('hello',name)
    t = threading.Thread(target=thread_run,)
    t.start()


if __name__=="__main__":
    for i in range(10):
        p = Process(target=func,args=('boy_%s'%i,))
        p.start()
        p.join()

example:要显示所涉及的各个进程ID

#要显示所涉及的各个进程ID
　　#每一个子进程都有一个父进程，都是有父进程启动


import multiprocessing
import os


def info(title):
    print(title)
    print('module name:',__name__)
    print('parent process:',os.getppid()) #获取父进程ID
    print('process id:',os.getpid()) #获取自身进程ID
    print('

')


def func(name):
    info('33[31;1m 启动子进程 func--33[0m')
    print('hello',name)

if __name__=="__main__":
    info("33[32;1m 启动父进程33[0m")
    p = multiprocessing.Process(target=func,args=('boy',))
    p.start()

"""
############执行结果################
 启动父进程
module name: __main__
parent process: 6140    #这个进程是pychram的进程
process id: 8072


 启动子进程 func--
module name: __mp_main__
parent process: 8072
process id: 8112

hello boy

"""

3.2 进程间通讯

两个进程之间的内存是独立的，是不共享的。父进程起了子进程，子进程就独立了；要想实现两个进程间的数据交换必须通过中间件：翻译。

方式一、Queues 实现数据的传递

使用方法跟threading里的queue差不多。（线程queue，只在同一进程下的线程间共享数据，不与其他进程下的线程共享数据；线程queue不能把数据传给其他的进程）

 1 #推荐的实现方式

　　from  multiprocessing import Queue,Process
 2 
 3 def func(k):
 4    k.put([42,None,'hello'])
 5 
 6 
 7 if __name__ == '__main__':
 8         k = Queue() #进程Q
 9         p = Process(target=func,args=(k,)) #克隆了一份k给子进程，不是共享Q
10         p.start()
11         print(k.get())

方式二、Pipes

Pipe（）函数返回一个由管道连接的连接对象，默认情况下是双工（双向），实现的是数据的传递。

#pipe管道实现方式，类似socket发送和接收消息
from multiprocessing import Process,Pipe

def func(conn):
    conn.send([42, None, 'hello from child'])  #子进程发送消息
    conn.send([42, None, 'hello from child'])
    print("", conn.recv())  #子进程接收消息
    conn.close()


if __name__=='__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=func, args=(child_conn,))
    p.start()
    print("parent", parent_conn.recv())  # 父进程接收信息：prints "[42, None, 'hello']"
    print("parent", parent_conn.recv())  # prints "[42, None, 'hello']"
    parent_conn.send("from parent")   #父进程发送消息
    p.join()

from multiprocessing import Process, Pipe


def f(conn):
    conn.send([42, None, 'hello from child'])
    conn.send([42, None, 'hello from child2'])
    print("from parent:",conn.recv())
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print(parent_conn.recv())  # prints "[42, None, 'hello']"
    print(parent_conn.recv())  # prints "[42, None, 'hello']"
    parent_conn.send("pf42280可好") # prints "[42, None, 'hello']"
    p.join()

方式三、Managers

Manager（）返回的管理器对象控制一个服务器进程，该进程保存Python对象并允许其他进程使用代理操作它们。
Manager（）返回的管理器将支持类型列表，dict，Namespace，Lock，RLock，Semaphore，BoundedSemaphore，Condition，Event，Barrier，Queue，Value和Array。

例如：

#实现共享数据，不再是数据的传递
from multiprocessing import Process,Manager
import os

def func(dict1,list1):
    dict1[os.getpid()] = os.getpid()
    list1.append(os.getpid())
    print(list1)

if __name__=="__main__":
    with Manager() as manager:  # 类似打开文件的with open() as f
        dict1 = manager.dict()  #生成一个字典，可以在多个进程之间共享和传递
        list1 = manager.list(range(5)) #生成一个列表，可以在多个进程之间共享和传递，默认有5个值
        p_list = []  #用于存放进程列表
        for i in range(10):
            p = Process(target=func,args=(dict1,list1))  #生成进程
            p.start()       #启动进程
            p_list.append(p)   #把进程添加到p_list列表中
        for res in p_list:
            res.join()   #等待所有线程结束
        print(dict1)
        print(list1)


'''
结果：
{6456: 6456, 7528: 7528, 7392: 7392, 3588: 3588, 7744: 7744, 6448: 6448, 4552: 4552, 6776: 6776, 7676: 7676, 7560: 7560}
[0, 1, 2, 3, 4, 6456, 7528, 7392, 3588, 7744, 6448, 4552, 6776, 7676, 7560]

'''

from multiprocessing import Process, Manager
import os

def f(d, l):
    d[1] = '1'
    d['2'] = 2
    d["pid%s" %os.getpid()] = os.getpid()
    l.append(1)
    print(l,d)


if __name__ == '__main__':
    with Manager() as manager:
        d = manager.dict()

        l = manager.list(range(5))

        p_list = []
        for i in range(10):
            p = Process(target=f, args=(d, l))
            p.start()
            p_list.append(p)
        for res in p_list:
            res.join()
        l.append("from parent")
        print(d)
        print(l)

3.3进程池

3.3.1进程同步，进程锁

如果不使用来自不同进程的锁定输出，则可能会混淆不清

from multiprocessing import Process,Lock
lock = Lock()

def func(l,i):
    l.acquire()
    try:
        print("hello world",i)
    finally:
        l.release()


if __name__=="__main__":
    for num in range(10):
        p = Process(target=func,args=(lock,num))
        p.start()

3.3.2进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进进程，那么程序就会等待，直到进程池中有可用进程为止。
进程池中有两个方法：
apply   ---同步执行，串行
apply_async  ---异步执行，并行

"""
apply   ---同步执行，串行
apply_async  ---异步执行，并行


注意
1、windose起线程池，一定要用if __name__ =="__main__",写在测试代码里
2、pool.close 与poo1.join的顺序，先pool.close，再poo1.join
3、一定要poo1.join，否则程序直接退出

"""

from multiprocessing import Process,Pool,freeze_support
import time,os

def Foo(i):
    time.sleep(2)
    print("子进程 in process",os.getpid())
    return i+100
#回调函数，可以干很多事，比如备份数据，备份完成后会写日志
def Bar(arg):
    print('---->exec done:',arg,os.getpid())

if __name__=="__main__":
    freeze_support()
    pool = Pool(processes=3) #允许进程池同时放入3个进程
    print("主进程",os.getpid())
    for i in range(10):  #进程池起10个进程，但同时只有3个（由pool = Pool(processes=3)限定）
        pool.apply_async(func=Foo,args=(i,),callback=Bar)  #并行    #callback=回调,由主进程完成
        #pool.apply(func=Foo,args=(i,)) #串行，进入进程池中的3个，一个一个的执行
        #pool.apply_async(func=Foo,args=(i,))#并行  进入进程池中的3个并行执行
    print('end')
    pool.close()  #先关闭进程池，再join
    pool.join()   #进程池中进程执行完毕后再关闭，如果注释，主进程完成后，直接关闭

3.3.3apply()和apply_async()的区别

apply()：
apply是阻塞的。首先主进程开始运行，碰到子进程，操作系统切换到子进程，等待子进程运行结束后，在切换到另外一个子进程，直到所有子进程运行完毕。然后在切换到主进程，运行剩余的部分。这样跟单进程串行执行没什么区别。
如：

import time
from multiprocessing import Pool

def run(count):
    print('子进程编号:%s'%count)
    time.sleep(2)
    print('子进程%s结束'%count)

if __name__=="__main__":
    print('开始执行主程序')
    star_time = time.time()
    #使用进程池创建子进程
    pool = Pool(3)
    print("开始执行子进程")
    for i in range(10):
        pool.apply(run,(i,))
    pool.close()
    pool.join()
    print("主进程结束，总耗时%s" %(time.time() - star_time))



"""
开始执行主程序
开始执行子进程
子进程编号:0
子进程0结束
子进程编号:1
子进程1结束
子进程编号:2
子进程2结束
子进程编号:3
子进程3结束
子进程编号:4
子进程4结束
子进程编号:5
子进程5结束
子进程编号:6
子进程6结束
子进程编号:7
子进程7结束
子进程编号:8
子进程8结束
子进程编号:9
子进程9结束
主进程结束，总耗时20.345163822174072

"""

可以看到子进程是顺序执行的，且子进程全部执行完毕后才继续执行主进程

apply_async()：
apply_async 是异步非阻塞的。即不用等待当前进程执行完毕，随时根据系统调度来进行进程切换。首先主进程开始运行，碰到子进程后，主进程仍可以先运行，等到操作系统进行进程切换的时候，在交给子进程运行。可以做到不等待子进程执行完毕，主进程就已经执行完毕，并退出程序。
如：

import time
from multiprocessing import Pool

def run(count):
    print('子进程编号:%s'%count)
    time.sleep(2)
    print('子进程%s结束'%count)

if __name__=="__main__":
    print('开始执行主程序')
    star_time = time.time()
    #使用进程池创建子进程
    pool = Pool(3)
    print("开始执行子进程")
    for i in range(10):
        pool.apply_async(run,(i,))
    pool.close()
    pool.join()
    print("主进程结束，总耗时%s" %(time.time() - star_time))

"""
进程的切换是操作系统来控制的，是抢占式的切换。 我们首先运行的是主进程，由于主进程代码很简单，主进程一下子就运行完毕了，所以子进程完全没有机会切换到程序就已经结束了。
如果我们想要子进程执行完毕后再运行主进程剩余部分，则在恰当位置上加上一句子进程名.join()。这样就告诉主进程等该子进程执行完毕后，再运行主进程剩余部分


开始执行主程序
开始执行子进程
子进程编号:0
子进程编号:1
子进程编号:2
子进程0结束
子进程编号:3
子进程1结束
子进程编号:4
子进程2结束
子进程编号:5
子进程3结束
子进程编号:6
子进程4结束
子进程编号:7
子进程5结束
子进程编号:8
子进程6结束
子进程编号:9
子进程7结束
子进程8结束
子进程9结束
主进程结束，总耗时8.29747462272644

"""

 可以看到子进程是并行执行的