day36,多线程,线程中的锁机制,joinablequeue

多线程

什么是线程 :是真正的执行单位

线程不能单独存在必须存在于进程中,

进程是一个资源单位,其包含了运行程序所需的所有资源

线程才是真正的执行单位

没有线程,进程中的资源无法被利用起来,所以一个进程至少包含一个线程,称之为主线程

当我们启动一个程序时,操作系统就会自己为这个程序创建一个主线程

线程可以由程序后期开启 ,自己开启线程称之为子线程

为什么需要线程:

它的目的只有一个就是提高效率,

因为在是在进程中,只有一个线程的话效率很低,多个线程就可以完成更多的工作,

还有一个原因就是,在创建多个进程来完成的话,就会及其的消耗资源

如何使用创建线程:

两种方法

一个使用Threading 类中的Thread 创建线程

第二个方法是继承Thread这个类覆盖run方法创建线程,这个和进程的创建是差不多的

实例:第一种方法

from threading import Thread,current_thread
import time

def task():
    print("2",current_thread())
    print("子线程running")
    time.sleep(10)
    print("子线程over")

# 使用方法一  直接实例化Thread类
if __name__ == '__main__':
    t = Thread(target=task)
    t.start()
    
    # task()
    # 执行顺序不固定 如果开启线程速度足够快  可能子线程先执行
    print("主线程over")
    print("1",current_thread())

实例:第二种方法

# 使用方法二 继承Thread 覆盖run方法
class MyThread(Thread):
    def run(self):
        print("子线程run!")
m = MyThread()
print("主线over")

# 使用方法和多进程一模一样   开启线程的代码可以放在任何位置  开启进程必须放在判断下面

线程的几个特点:
1 开启线程不消耗资源,运行速度比叫进程要快

2 在开启多个子线程后,线程之间的数据是共享的,所以就不存在什么进程中的queue和Manager

3 线程中是没有父子级的,都是相同的pid不存在谁要先执行,后执行的,都是通过自己抢占的,就算是有父子级,都是自己设置的

守护线程:

守护线程是和进程中的守护进程是一样的,就是被守护的线程如果死掉的话,那么守护线程也会随着死掉

JoinAbleQueue

这是一个相当一是Queue这样的一个列队,joinablequeue

不同的是队列允许项目的使用者通知生成者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。

joinablequeue 中间需要注意的是在获取数据是需要使用 q.task_done()这个方法告诉进程,获取的每一个值,知道列队中的值被取完

就算值取完了,列队也是会等待在添加值,然后在取,这样进程是不会结束的,所以就把这个进程设置为守护进程,通过被守护的进程结束

那么,守护进程也会结束,看这样整个程序就结束了

JoinAbleQueue 实例:

from multiprocessing import Process,JoinableQueue
import time,random
def consumer(q):
    while True:
        time.sleep(random.randint(1,5))
        res=q.get()
        print('消费者拿到了 %s' %res)
        q.task_done()

def producer(seq,q):
    for item in seq:
        time.sleep(random.randrange(1,2))
        q.put(item)
        print('生产者做好了 %s' %item)
    q.join()

if __name__ == '__main__':
    q=JoinableQueue()
    seq=('包子%s' %i for i in range(10))
    p=Process(target=consumer,args=(q,))
    p.daemon=True #设置为守护进程，在主线程停止时p也停止，但是不用担心，producer内调用q.join保证了consumer已经处理完队列中的所有元素
    p.start()

    producer(seq,q)

    print('主线程')

线程中的锁

共享意味着竞争

线程中也存在安全问题,

多线程可以并发执行,一旦并发了并且访问了同一个资源就会有问题

解决方案:还是互斥锁

from threading import Thread,enumerate,Lock
import time

number = 10

lock = Lock()

def task():
    global number
    lock.acquire()
    a = number
    time.sleep(0.1)
    number = a - 1
    lock.release()

for i in range(10):
    t = Thread(target=task)
    t.start()

for t in enumerate()[1:]:
    # print(t)
    t.join()
    
print(number)

View Code

死锁:

死锁就是在有多个锁的多个线程中,一个资源需要多重锁来锁住资源,那么有多个线程抢占这个资源,就会发生一个线程抢占一个锁,

满足不了,这个资源需要双重锁,这样就会发生,死锁的问题,都不解不了锁,所以程序就只能等待下去

可重入锁

Rlock 同一个线程可以多次执行acquire，释放锁时，有几次acquire就要release几次。

但是本质上同一个线程多次执行acquire时没有任何意义的，其他线程必须等到RLock全部release之后才能访问共享资源。

所以Rlock仅仅是帮你解决了代码逻辑上的错误导致的死锁，并不能解决多个锁造成的死锁问题

Rlock 的实例:

# 同一把RLock 多次acquire
#l1 = RLock()
#l2 = l1

# 不同的RLock 依然会锁死
import threading
from threading import Thread
from multiprocessing import RLock

l1 = RLock()
l2 = RLock()

def task():
    l1.acquire()
    print(threading.current_thread().name,"拿到了筷子")
    l2.acquire()
    print(threading.current_thread().name, "拿到了盘子")

    print("吃饭")
    l1.release()
    l2.release()

def task2():
    l2.acquire()
    print(threading.current_thread().name, "拿到了盘子")

    l1.acquire()
    print(threading.current_thread().name,"拿到了筷子")

    print("吃饭")

    l2.release()
    l1.release()

t1 = Thread(target=task)
t1.start()
t2 = Thread(target=task2)
t2.start()

View Code

信号量:

Semaphore

信号量也是一种锁，其特殊之处在于可以让一个资源同时被多个线程共享，并控制最大的并发访问线程数量。

Semaphore 是用于用户访问服务器的最大的并发量,可以限制访问的人数,一般不用于线程

如果把Lock比喻为家用洗手间，同一时间只能一个人使用。

那信号量就可以看做公共卫生间，同一时间可以有多个人同时使用。

from threading import  Thread,Semaphore,current_thread
import time

s = Semaphore(3)
def task():
    s.acquire()
    print("%s running........" % current_thread())
    time.sleep(1)
    s.release()
    
for i in range(20):
    Thread(target=task).start()