python基础---多线程

时间 2019-11-06

标签 python 基础多线程栏目 Python 繁體版

原文原文链接

多线程
python

1 线程：mysql

线程顾名思义，就是一条流水线工做的过程，一条流水线必须属于一个车间，一个车间的工做过程是一个进程，车间负责把资源整合到一块儿，是一个资源单位，而一个车间内至少有一个流水线sql

进程只是用来把资源集中到一块儿（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位
安全

多线程：
服务器

多线程（即多个控制线程）的概念是，在一个进程中存在多个控制线程，多个控制线程共享该进程的地址空间，至关于一个车间内有多条流水线，都共用一个车间的资源多线程

1. 多线程共享一个进程的地址空间并发

2. 线程比进程更轻量级，线程比进程更容易建立可撤销，在许多操做系统中，建立一个线程比建立一个进程要快10-100倍，在有大量线程须要动态和快速修改时，这一特性颇有用app

3. 若多个线程都是cpu密集型的，那么并不能得到性能上的加强，可是若是存在大量的计算和大量的I/O处理，拥有多个线程容许这些活动彼此重叠运行，从而会加快程序执行的速度。dom

4. 在多cpu系统中，为了最大限度的利用多核，能够开启多个线程，比开进程开销要小的多。（这一条并不适用于python）ide

2 开启线程的两种方式

#开启线程的方式一：使用替换threading模块提供的Thread
from threading import Thread
from multiprocessing import Process

def task():
    print('is running')

if __name__ == '__main__':
    t=Thread(target=task,)
    # t=Process(target=task,)
    t.start()
    print('主')

#开启线程的方式二：自定义类，继承Thread
from threading import Thread
from multiprocessing import Process
class MyThread(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('%s is running' %self.name)

if __name__ == '__main__':
    t=MyThread('egon')
    # t=Process(target=task,)
    t.start()
    print('主')

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

from threading import Thread
from multiprocessing import Process
import os

def task():
    print('%s is running' %os.getpid())

if __name__ == '__main__':
    t1=Thread(target=task,)
    t2=Thread(target=task,)
    t1.start()
    t2.start()    # 在主进程下开启多个线程,每一个线程都跟主进程的pid同样
    print('主',os.getpid())
    t3=Process(target=task,)
    t4=Process(target=task,)
    t3.start()
    t4.start()    # 开多个进程,每一个进程都有不一样的pid
    print('主',os.getpid())

# 多线程共享统一进程里的资源
from threading import Thread
from multiprocessing import Process
n=100
def work():
    global n
    n=0

if __name__ == '__main__':

    p=Process(target=work,)
    p.start()
    p.join()
    print('主',n)

    t=Thread(target=work,)
    t.start()
    t.join()
    print('主',n)

4 守护进程

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

须要强调的是：运行完毕并不是终止运行

1.对主进程来讲，运行完毕指的是主进程代码运行完毕
2.对主线程来讲，运行完毕指的是主线程所在的进程内全部非守护线程通通运行完毕，主线程才算运行完毕

详细解释：

1 主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,而后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(不然会产生僵尸进程)，才会结束
2 主线程在其余非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。由于主线程的结束意味着进程的结束，进程总体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束

#先看：守护进程
#
from multiprocessing import Process
import time

def task1():
    print('123')
    time.sleep(1)
    print('123done')

def task2():
    print('456')
    time.sleep(10)
    print('456done')

if __name__ == '__main__':
    p1=Process(target=task1)
    p2=Process(target=task2)
    p1.daemon = True
    p1.start()
    p2.start()
    print('主')


#再看：守护线程

from threading import Thread
import time

def task1():
    print('123')
    time.sleep(10)
    print('123done')

def task2():
    print('456')
    time.sleep(1)
    print('456done')

if __name__ == '__main__':
    t1=Thread(target=task1)
    t2=Thread(target=task2)
    t1.daemon=True
    t1.start()
    t2.start()
    print('主')

5 同步锁（GIL锁）

须要注意的点：
1.线程抢的是GIL锁，GIL锁至关于执行权限，拿到执行权限后才能拿到互斥锁Lock，其余线程也能够抢到GIL，但若是发现Lock仍然没有被释放则阻塞，即使是拿到执行权限GIL也要马上交出来
2.join是等待全部，即总体串行，而锁只是锁住修改共享数据的部分，即部分串行，要想保证数据安全的根本原理在于让并发变成串行，join与互斥锁均可以实现，毫无疑问，互斥锁的部分串行效率要更高

机智的同窗可能会问到这个问题，就是既然你以前说过了，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早期版本的遗留问题

from threading import Thread
n=100
def task():
    print('is running')

if __name__ == '__main__':
    t1=Thread(target=task,)
    t2=Thread(target=task,)
    t3=Thread(target=task,)
    # t=Process(target=task,)
    t1.start()
    t2.start()
    t3.start()
    print('主')

锁一般被用来实现对共享资源的同步访问。为每个共享资源建立一个Lock对象，当你须要访问该资源时，调用acquire方法来获取锁对象（若是其它线程已经得到了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁

互斥锁

from threading import Thread,Lock
import time
n=100
def work():
    global n
    mutex.acquire()
    temp=n
    time.sleep(0.1)
    n=temp-1
    mutex.release()

if __name__ == '__main__':
    mutex=Lock()
    l=[]
    start=time.time()
    for i in range(100):
        t=Thread(target=work)
        l.append(t)
        t.start()

    for t in l:
        t.join()
    print('run time:%s value:%s' %(time.time()-start,n))

同步锁与互斥锁：

 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,Lockimport os,timedef task():    global n    print('%s is running' %current_thread().getName())
    temp=n
    time.sleep(0.5)
    n=temp-1if __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,Lockimport os,timedef 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,Lockimport os,timedef task():
    time.sleep(3)    print('%s start to run' %current_thread().getName())    global n
    temp=n
    time.sleep(0.5)
    n=temp-1if __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 #耗时是多么的恐怖'''

#多进程：
#优势：能够利用多核优点
#缺点:开销大


#多线程：
#优势：开销小
#缺点:不能利用多核优点

from threading import Thread
from multiprocessing import Process
import time
#计算密集型
def work():
    res=1
    for i in range(100000000):
        res+=i

if __name__ == '__main__':
    p_l=[]
    start=time.time()
    for i in range(4):
        # p=Process(target=work) #6.7473859786987305
        p=Thread(target=work) #24.466399431228638
        p_l.append(p)
        p.start()
    for p in p_l:
        p.join()

    print(time.time()-start)


from threading import Thread
from multiprocessing import Process
import time
#IO密集型
def work():
    time.sleep(2)

if __name__ == '__main__':
    p_l=[]
    start=time.time()
    for i in range(400):
        # p=Process(target=work) #12.104692220687866
        p=Thread(target=work) #2.038116455078125
        p_l.append(p)
        p.start()
    for p in p_l:
        p.join()

    print(time.time()-start)

6 死锁与递归锁

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

# 死锁现象
from threading import Thread,Lock,RLock
import time
mutexA=Lock()
mutexB=Lock()
class Mythread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutexA.acquire()
        print('\033[45m%s 抢到A锁\033[0m' %self.name)
        mutexB.acquire()
        print('\033[44m%s 抢到B锁\033[0m' %self.name)
        mutexB.release()
        mutexA.release()

    def f2(self):
        mutexB.acquire()
        print('\033[44m%s 抢到B锁\033[0m' %self.name)
        time.sleep(1)
        mutexA.acquire()
        print('\033[45m%s 抢到A锁\033[0m' %self.name)
        mutexA.release()
        mutexB.release()


if __name__ == '__main__':
    for i in range(20):
        t=Mythread()
        t.start()

#递归锁
from threading import Thread,Lock,RLock
import time
mutex=RLock()
class Mythread(Thread):
    def run(self):
        self.f1()
        self.f2()

    def f1(self):
        mutex.acquire()
        print('\033[45m%s 抢到A锁\033[0m' %self.name)
        mutex.acquire()
        print('\033[44m%s 抢到B锁\033[0m' %self.name)
        mutex.release()
        mutex.release()

    def f2(self):
        mutex.acquire()
        print('\033[44m%s 抢到B锁\033[0m' %self.name)
        time.sleep(1)
        mutex.acquire()
        print('\033[45m%s 抢到A锁\033[0m' %self.name)
        mutex.release()
        mutex.release()


if __name__ == '__main__':
    for i in range(20):
        t=Mythread()
        t.start()

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

这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源能够被屡次require。直到一个线程全部的acquire都被release，其余的线程才能得到资源。上面的例子若是使用RLock代替Lock，则不会发生死锁：

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

7 信号量Semaphore

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

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

sm=Semaphore(5)
def work():
    sm.acquire()
    print('%s 上厕所' %current_thread().getName())
    time.sleep(random.randint(1,3))
    sm.release()

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

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

8 事件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

例如，有多个工做线程尝试连接MySQL，咱们想要在连接前确保MySQL服务正常才让那些工做线程去链接MySQL服务器，若是链接不成功，都会去尝试从新链接。那么咱们就能够采用threading.Event机制来协调各个工做线程的链接操做

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

def conn_mysql():
    count=1
    while not event.is_set():
        if count > 3:
            raise ConnectionError('连接失败')
        print('%s 等待第%s次连接mysql' %(current_thread().getName(),count))
        event.wait(0.5)
        count+=1

    print('%s 连接ok' % current_thread().getName())


def check_mysql():
    print('%s 正在检查mysql状态' %current_thread().getName())
    time.sleep(1)
    event.set()


if __name__ == '__main__':
    t1=Thread(target=conn_mysql)
    t2=Thread(target=conn_mysql)
    check=Thread(target=check_mysql)

    t1.start()
    t2.start()
    check.start()

9 定时器

from threading import Timer


def hello(n):
    print("hello, world",n)


t = Timer(3, hello,args=(11,))
t.start()  # after 1 seconds, "hello, world" will be printed

10 线程queue

import queue

q=queue.Queue(3) #队列：先进先出
q.put(1)
q.put(2)
q.put(3)

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


q=queue.LifoQueue(3) #堆栈：后进先出
q.put(1)
q.put(2)
q.put(3)

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


q=queue.PriorityQueue(3) #数字越小优先级越高
q.put((10,'data1'))
q.put((11,'data2'))
q.put((9,'data3'))

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

待整理。。。