并发编程-多线程

threading模块开启线程的两种方式

# 方式一

from threading import Thread

def piao(name):
    print('%s piaoing'%name)
    time.sleep(random.randrange(1,5))
    print('%s piao end'%name)
if __name__ == '__main__':
    t1=Thread(target=piao,args=('egon',))
    t1.start()
    print('主线程')

#方式二
import time
import random
from threading import Thread

class Piao(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('%s piaoing' %self.name)
        time.sleep(random.randrange(1,5))
        print('%s piao end'%self.name)

if __name__ == '__main__':
    t1=Piao('egon')
    t1.start()
    print('主')

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

1.开进程的开销 远大于开线程

2.同一进程内的多个线程共享该进程的地址空间

3.一个进程下开启多个线程的pid相同，一个进程下开启多个子进程pid不同。

1、开进程的开销 远大于开进程
import time
from threading import Thread
from multiprocessing import Process

def Piao(name):
    print('%s is piaoing'%name)
    time.sleep(1)
    print('%s piao end'%name)

if __name__ == '__main__':
    # p1=Process(target=Piao,args=('egon',))
    # p1.start()
    t1=Thread(target=Piao,args=('egon',))
    t1.start()
    print('主线程')

2、同一进程内的多个线程共享该进程的地址空间

from threading import Thread
from multiprocessing import Process

n=100
def task():
    global n
    n = 0

if __name__ == '__main__':
    # p1=Process(target=task,)
    # p1.start()
    # p1.join()
    t1=Thread(target=task,)
    t1.start()
    t1.join()
    print('主线程',n)

3、看一眼 pid
from multiprocessing import Process,current_process
import os

def task():
    # print(current_process().pid)  # 方式一
    print('子进程pid:%s  父进程的pid:%s'%(os.getpid(),os.getppid()))  #方式二

if __name__ == '__main__':
    p1=Process(target=task,)
    p1.start()

    # t1=Thread(target=task,)
    # t1.start()

    print('主线程',current_process().pid)

线程相关的其他方法

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

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

from threading import Thread,currentThread,active_count
import time

def task():
    print('%s is running'%currentThread().getName())  # 子线程的名字
    time.sleep(1)
    print('%s is done'%currentThread().getName())
if __name__ == '__main__':
    t=Thread(target=task,name='子线程1')
    t.start()
    # t.setName('儿子线程1')  # 给子线程重新命名为 儿子线程1
    # t.join()   #主线程等子线程运行结束后
    # currentThread().setName('主线程')  # 给主线程改名字
    # print(t.isAlive())

    # print('主线程',currentThread().getName() )  # 主线程的名字

    # t.join()
    # print(active_count())  # 查看活跃数
    print(enumerate())  #连同主线程在内有两个运行的线程

守护线程 

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

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

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) # 设置方法一，要在start 前设置
    t.daemon=True  #设置守护线程或者守护进程，要在start开始前设置
    t.start()

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


from threading import Thread
import time

def foo():
    print(123)
    time.sleep(1)
    print("end123")

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

if __name__ == '__main__':
    t1=Thread(target=foo)
    t2=Thread(target=bar)

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

#运行结果：
"""
123
456
main-------
end123
end456
"""
#分析：t1是守护线程，t2是非守护线程，print("main------"是主线程)
#主线程 需要等子线程运行结束，t1是守护主线程的，主线程运行结束，守护线程就会结束

GIL互斥锁介绍

GIL本质就是一把互斥锁，既然是互斥锁，所有互斥锁的本质都一样，都是将并发运行变成串行，以此来控制同一时间内共享数据只能被一个任务所修改，进而保证数据安全。

GIL保护的是解释器级的数据，保护用户自己的数据则需要自己加锁处理

GIL 与Lock是两把锁，保护的数据不一样，前者是解释器级别的（当然保护的就是解释器级别的数据，比如垃圾回收的数据），后者是保护用户自己开发的应用程序的数据，很明显GIL不负责这件事，只能用户自定义加锁处理，即Lock

join与加锁的区别

不加锁:并发执行,速度快,数据不安全

不加锁:未加锁部分并发执行,加锁部分串行执行,速度慢,数据安全

start后立即join:任务内的所有代码都是串行执行的,而加锁,只是加锁的部分即修改共享数据的部分是串行的 单从保证数据安全方面,二者都可以实现,但很明显是加锁的效率更高.

死锁现象与递归锁

进程也有死锁与递归锁，在进程那里忘记说了，放到这里一切说了额

所谓死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程，如下就是死锁

# 死锁：因为互斥锁没有及时释放，造成的死锁。
from threading import Thread,Lock

mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到A锁' %self.name)

        mutexB.acquire()
        print('%s拿到B锁'% self.name)
        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('%s 拿到B锁'%self.name)
        time.sleep(0.1)

        mutexA.acquire()
        print('%s 拿到A锁'%self.name)
        mutexA.release()
        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()

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

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁：

#递归锁：可以连续acquire多次，每一次acquire一次计算器加1，只有计数器为0时，才能被抢到acquire.
from threading import Thread,RLock
import time

mutexB=mutexA=RLock()

class MyThread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('%s 拿到A锁' %self.name)

        mutexB.acquire()
        print('%s拿到B锁'% self.name)
        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('%s 拿到B锁'%self.name)
        time.sleep(0.1)

        mutexA.acquire()
        print('%s 拿到A锁'%self.name)
        mutexA.release()
        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()

信号量Semaphore

同进程的一样

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

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

from threading import Thread,Semaphore,currentThread
import random,time

sm=Semaphore(3)

def task():
    # sm.acquire()
    # print('%s in'%currentThread().getName())
    # sm.release()
    with sm:
        print('%s in'%currentThread().getName())
        time.sleep(random.randint(1,3))



if __name__ == '__main__':
    for i in range(10):
        t=Thread(target=task)
        t.start()

Event事件

threading库中的Event对象:其 他线程需要通过判断某个线程的状态来确定自己下一步的操作

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

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

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

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

from threading import Thread,Event
import time
event=Event()

def student(name):
    print('%s 在上课'%name)
    event.wait(2)   # 超时间就可以出去
    print('%s 课间活动'%name)

def teacher(name):
    print('%s老师在授课'%name)
    time.sleep(4)
    event.set()

if __name__ == '__main__':
    s1=Thread(target=student,args=('alex',))
    s2=Thread(target=student,args=('wxx',))
    s3=Thread(target=student,args=('yxx',))
    t1=Thread(target=teacher,args=('egon',))

    s1.start()
    s2.start()
    s3.start()
    t1.start()

from threading import Thread, Event, currentThread
import time

event = Event()


def conn():
    n = 0
    while not event.is_set():
        if n == 3:
            print('%s 连接超时，请重新连接'%currentThread().getName())
            return
        print('%s try is conning' % currentThread().getName())
        event.wait(0.5)
        n += 1
    print('%s is connected' % currentThread().getName())


def check():
    print('%s is checking' % currentThread().getName())
    time.sleep(5)
    event.set()


if __name__ == '__main__':
    for i in range(3):
        t = Thread(target=conn)
        t.start()
    t = Thread(target=check)
    t.start()

#3个正在连接，4在检测，检测完后，才能连接成功。

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

from threading import Timer

def task(name):
    print('%s  hello'%name)

t=Timer(5,task,args=('egon',))
t.start()


#制作随机验证码
from threading import Timer
import random

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

    def make_cache(self,interval=10): #默认时间间隔3秒
        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:
            code=input('>>:').strip()
            if code.upper() == self.cache:
                print('验证码正确')
                self.t.cancel()
                break


obj=Code()
obj.check()

线程queue

queue队列 ：使用import queue，用法与进程Queue一样

import queue
q=queue.Queue(3)  # 先进先出-->队列,队列大小为3
q.put('first')  # 一次一次放值，如果放满了，就放不下了，会锁掉
q.put(2)
q.put('third')

q.put(4,block=True) #block 默认是TRUE,是阻塞状态
q.put(4,block=False) #修改为Flase 是不阻塞的
q.put(4,block=True,timeout=3) #block 默认是TRUE,是阻塞状态,阻塞时间3秒

print(q.get())  # 一次一次取值
print(q.get())
print(q.get())

print(q.get(block=True))  #无数据，会阻塞
print(q.get(block=False))  #无数据，不阻塞，但是会报错
print(q.get_nowait())  # 不等待
print(q.get(block=True,timeout=3))


q=queue.LifoQueue(3)  # 后进先出--》堆栈
q.put('first')  # 一次一次放值，如果放满了，就放不下了，会锁掉
q.put(2)
q.put('third')
print(q.get())  # 一次一次取值
print(q.get())
print(q.get())


q=queue.PriorityQueue(3)  #优先级:数字越小，优先级越高，会先拿出来
q.put((10,'one'))
q.put((40,'two'))
q.put((30,'three'))

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

多线程实现并发的套接字

from socket import *
from concurrent.futures import ThreadPoolExecutor
#整理成函数的优点：将功能分清楚

def commumicate(conn):
    while True:
        try:
            data = conn.recv(1024)
            print(data)
            if not data: break
            conn.send(data.upper())
        except ConnectionRefusedError:
            break
    conn.close()


def server(ip, port):
    server = socket(AF_INET, SOCK_STREAM)
    server.bind((ip,port))
    server.listen(5)

    while True:
        conn, addr = server.accept()
        pool.submit(commumicate,conn)

    server.close()

if __name__ == '__main__':
    pool=ThreadPoolExecutor(2)  #设定线程池，可以接受的最大线程数
    server('127.0.0.1',8080)

客户端

from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))

while True:
    cmd = input('>>:').strip()
    if not cmd:continue
    client.send(cmd.encode('utf-8'))
    data = client.recv(1024)
    print(data.decode('utf-8'))

client.close()

进程池与线程池

为控制服务开启的进程或者线程数会随着客户端数目的增多而增多而带来的压力，让机器在自己可以承受的范围，我们使用进程池或线程池。

from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
import os,time,random
def task(name):
    print('name:%s run %s'%(name,os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == '__main__':
    pool=ThreadPoolExecutor(5)  #线程池
    # pool=ProcessPoolExecutor(4)  #进程池
    for i in range(10):
        pool.submit(task,'egon %s'%i)

    pool.shutdown(wait=True)  #

    print('zhu')


from concurrent.futures import ProcessPoolExecutor,ThreadPoolExecutor
from threading import currentThread
import os,time,random
def task():
    print('name:%s run %s'%(currentThread().getName(),os.getpid()))
    time.sleep(random.randint(1,3))

if __name__ == '__main__':
    pool=ThreadPoolExecutor(5)  #线程池
    # pool=ProcessPoolExecutor(4)  #进程池
    for i in range(10):
        pool.submit(task,)

    pool.shutdown(wait=True)  #

    print('zhu')

异步调用与回调机制

提交任务的两种方式
 1.同步调用:提交任务后，在原地等待任务执行完毕，拿到结果，在执行下一行代码；导致程序串行执行

from concurrent.futures import ThreadPoolExecutor
import time
import random

def la(name):
    print('%s is laing'%name)
    time.sleep(random.randint(3,5))
    res=random.randint(7,13)*'#'
    return {'name':name,'res':res}
def weight(shit):
    name=shit['name']
    size=len(shit['res'])
    print('%s 拉了 <%s>kg'%(name,size))

if __name__ == '__main__':
    pool=ThreadPoolExecutor(13)
    shit1=pool.submit(la,'alex').result()  # 加上result能得到结果
    weight(shit1)
    shit2=pool.submit(la,'wupiqi').result()
    weight(shit2)
    shit3=pool.submit(la,'yuanhao').result()
    weight(shit3)

异步调用：提交任务后，不在等待任务执行完毕

from concurrent.futures import ThreadPoolExecutor
import time
import random

def la(name):
    print('%s is laing'%name)
    time.sleep(random.randint(3,5))
    res=random.randint(7,13)*'#'
    return {'name':name,'res':res}

def weight(shit):
    shit=shit.result()
    name=shit['name']
    size=len(shit['res'])
    print('%s 拉了 <%s>kg'%(name,size))

if __name__ == '__main__':
    pool=ThreadPoolExecutor(13)
    pool.submit(la,'alex').add_done_callback(weight)

    pool.submit(la,'wupiqi').add_done_callback(weight)

    pool.submit(la,'yuanhao').add_done_callback(weight)

小练习：

异步调用 加上回调机制  是在爬取数据来用的

from concurrent.futures import ThreadPoolExecutor
import requests
import time

def get(url):
    print('GET %s'%url)
    respone = requests.get(url)  #requests.get下载内容
    time.sleep(3)
    return {'url':url,'content':respone.text}

def parse(res):  # 解析
    res = res.result()
    print('%s parse res is %s'%(res['url'],len(res['content'])))

if __name__ == '__main__':
    urls=[
        'https://www.luffycity.com/micro/play/1263',
        'https://www.cnblogs.com/linhaifeng/p/7278389.html',
        'https://www.cnblogs.com/linhaifeng/p/7278389.html'
    ]

    pool=ThreadPoolExecutor(2)

    for url in urls:
        pool.submit(get,url).add_done_callback(parse)

回调函数：

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程：我好了额，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

map用法：

from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor

import os,time,random
def task(n):
    print('%s is runing' %os.getpid())
    time.sleep(random.randint(1,3))
    return n**2

if __name__ == '__main__':

    executor=ThreadPoolExecutor(max_workers=3)

    # for i in range(11):
    #     future=executor.submit(task,i)

    executor.map(task,range(1,12)) #map取代了for+submit