码迷，mamicode.com
首页 > 编程语言 > 详细

多线程

时间：2020-10-26 11:42:31      阅读：85      评论：0      收藏：0      [点我收藏+]

标签：count   lazy   false   图片   inf   png   释放   cti   finish   

　　

from threading import Event,Thread
import logging,time

FORMAT=‘%(asctime)s %(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)

def boss(event:Event):
    logging.info(‘boss is waitting‘)
    event.wait()
    logging.info(‘boss: work is done‘)

def worker(event:Event,count=10):
    logging.info(‘worker start‘)
    cups=[]
    while True:
        cups.append(1)
        time.sleep(0.5)
        logging.info(‘append 1‘)
        if len(cups)>=10:
            event.set()
            break
    logging.info(‘worker finished, cups={}‘.format(cups))

event=Event()
w=Thread(target=worker,args=(event,))
b=Thread(target=boss,args=(event,))
w.start()
b.start()

 

使用同一个Event对象的标记flag

event.wait()就是等到flag变为True,或等到超时返回False,不限制等待时间

 

from threading import Event,Thread
import logging,time
logging.basicConfig(level=logging.INFO)

def do(event:Event,interval:int):
    while not event.wait(interval):
        logging.info(‘do sth‘)

e=Event()
Thread(target=do,args=(e,3)).start()

# e.wait(10)
time.sleep(10)
e.set()
print(‘main exited‘)

wait优于sleep,wait会主动出让时间片,其它线程可以调度,而sleep会占用时间片,不出让

 

Timer类的实现

from threading import Thread,Event
import logging,datetime
logging.basicConfig(level=logging.INFO)

def add(x:int,y:int):
    logging.info(x+y)

class Timer():
    def __init__(self,interval,fn,*args,**kwargs):
        self.interval=interval
        self.fn=fn
        self.args=args
        self.kwargs=kwargs
        self.event=Event()

    def start(self):
        Thread(target=self.__run).start()

    def cancel(self):
        self.event.set()

    def __run(self):
        start=datetime.datetime.now()
        logging.info(‘waiting‘)

        self.event.wait(self.interval)
        if not self.event.is_set():
            self.fn(*self.args,**self.kwargs)

        delta=(datetime.datetime.now()-start).total_seconds()
        logging.info(‘finished {}‘.format(delta))

t=Timer(5,add,3,9)
t.start()
e=Event()
e.wait(3)
t.cancel()

 

Lock:

lock 存在共享资源争抢的地方都可以使用索,从而保证只有一个使用者可以完全使用这个资源

import logging,threading
logging.basicConfig(level=logging.INFO)
cups=[]
lock=threading.Lock()

def w(lock:threading.Lock,n=100):
    while True:
        lock.acquire()
        count=len(cups)
        logging.info(count)
        if count>=n:
            lock.release()
            break
        cups.append(1)
        lock.release()
        logging.info(‘{} produce‘.format(threading.current_thread()))


for b in range(10):
    threading.Thread(target=w,args=(lock,20)).start()

threading.Event().wait(1)
print(len(cups))
print(threading.enumerate())

 

import threading,logging,time
FORMAT=‘%(name)s %(asctime)s %(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups=[]

def worker(count=10):
    logging.info(‘produce cups‘)
    flag=True
    while True:
        if len(cups)>=count:
            flag=False
        time.sleep(0.001)
        if not flag:
            break
        if flag:
            cups.append(1)
    logging.info(‘{} done,cups: {}‘.format(threading.current_thread(),len(cups)))

for _ in range(10):
    threading.Thread(target=worker,args=(20,)).start()
threading.Event().wait(1)
print(len(cups))

 

Lock

一旦线程获得锁,其它试图获得锁的线程将被 阻塞,

acquire(blocking=True,timeout=-1) 默认阻塞,可以设置阻塞超时时间,非阻塞时,timeout禁止设置,成功获取锁,返回True,否则返回False
release() 释放锁,可以从任何线程调用release(),已上锁的Lock会被重置为unlocked,未上锁调用release(),抛出RuntimeError异常

 

 

Counter类

import threading,time
class C:
    def __init__(self):
        self.__val=0
    @property
    def value(self):
        return self.__val
    def inc(self):
        lock.acquire()
        self.__val+=1
        lock.release()
    def dec(self):
        lock.acquire()
        self.__val-=1
        lock.release()
lock=threading.Lock()
def run(c:C,count=100):
    for _ in range(count):
        for p in range(-50,50):
            if p<0:
                # lock.acquire()
                c.dec()
                # lock.release()
            else:
                # lock.acquire()
                c.inc()
                # lock.release()
c=C()
c1=10
c2=10000
for b in range(c1):
    threading.Thread(target=run,args=(c,c2)).start()
# time.sleep(0.001)
print(c.value)
while True:
    time.sleep(0.5)
    print(threading.enumerate())
    print(c.value)

类中加锁 或者 操作加锁

 

通过with 或 try finally确保锁释放

import threading,time

class C:
    def __init__(self):
        self.__val=0
        self.__lock=threading.Lock()
    @property
    def value(self):
        with self.__lock:
            return self.__val
    def inc(self):
        try:
            self.__lock.acquire()
            self.__val+=1
        finally:
            self.__lock.release()
    def dec(self):
        try:
            self.__lock.acquire()
            self.__val-=1
        finally:
            self.__lock.release()
    def dec(self):
        with self.__lock:
            self.__val-=1
def run(c:C,count=100):
    for _ in range(count):
        for p in range(-50,50):
            if p<0:
                c.inc()
            else:
                c.dec()
c=C()
c1=10
c2=1000
for _ in range(c1):
    threading.Thread(target=run,args=(c,c2)).start()

while True:
    if threading.active_count() == 1:
        print(threading.enumerate())
        print(c.value)
    else:
        print(threading.enumerate())
    time.sleep(0.5)

 

import threading,logging,time
FORMAT=‘%(name)s %(asctime)s %(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups=[]

def worker(lock:threading.Lock,task=100):
    while True:
        if lock.acquire(blocking=False):
            count=len(cups)
            logging.info(‘number of cups: {}‘.format(count))

            if count>=task: # 每个子线程必须执行此退出条件
                lock.release()
                break

            cups.append(1)
            lock.release()
            logging.info(‘{} produce {} cup‘.format(threading.current_thread().name,len(cups)))
            time.sleep(0.5)
    logging.info(‘worker end,number of cups: {}‘.format(len(cups)))

lock=threading.Lock()
for _ in range(10):
    threading.Thread(target=worker,args=(lock,40)).start()

while True:
    if threading.active_count() == 1:
        print(threading.enumerate())
        print(len(cups))
    else:
        print(threading.enumerate())
    time.sleep(1)
import threading,logging,time
FORMAT=‘%(asctime)-15s\t[%(threadName)s %(thread)8d] %(message)s‘
logging.basicConfig(level=logging.INFO,format=FORMAT)

def worker(tasks):
    for task in tasks:
        time.sleep(0.01) # 强行切换线程,使任务依次执行
        if task.lock.acquire(blocking=False):
            logging.info(‘{} {} acquire success‘.format(threading.current_thread(),task.name))
        else:
            logging.info(‘{} {} acquire failed‘.format(threading.current_thread(),task.name))

class Task:
    def __init__(self,name):
        self.name=name
        self.lock=threading.Lock()

tasks=[Task(‘task-{}‘.format(p)) for p in range(10)]

for b in range(5):
    threading.Thread(target=worker,name=‘worker-{}‘.format(b),args=(tasks,)).start()

 

可重入锁RLock          Reentrant Lock

线程相关锁

线程A获得可重入锁,并可以多次获取成功,不会阻塞,最后要在线程A中做和acquire()次数相同的release()

import threading
lock=threading.RLock()
ret=lock.acquire()
print(ret)
ret=lock.acquire(timeout=5)
print(ret)
print(lock.acquire(True))
print(lock.acquire(False))

技术图片

 

 

import threading
def sub(lock:threading.RLock):
    lock.release()

lock=threading.RLock()
print(lock.acquire())
print(lock.acquire(timeout=3))

threading.Thread(target=sub,args=(lock,)).start()

技术图片

 

 

import threading,time
lock=threading.RLock()
print(lock.acquire())
print(‘~‘*80)
print(lock.acquire(blocking=True))
print(lock.acquire(blocking=False))
print(lock.acquire(timeout=3))
# for _ in range(5): # RuntimeError
#     lock.release()
# print(lock.acquire(blocking=False,timeout=3)) # ValueError

for _ in range(4): # 使Timer线程不阻塞
    lock.release()
def sub(lock:threading.RLock):
    print(‘{}: {}‘.format(threading.current_thread(),lock.acquire()))
    print(‘{}: {}‘.format(threading.current_thread(),lock.acquire()))
    print(lock.release())
    print(lock.release())
    # print(lock.release())

threading.Timer(2,sub,args=(lock,)).start()

不同线程获取同一个RLock时会阻塞

 

threading.Condition:

import threading,logging,random
logging.basicConfig(level=logging.INFO,format=‘%(asctime)s %(threadName)s %(thread)d %(message)s‘)

class Dispatcher:
    def __init__(self):
        self.data=0
        self.event=threading.Event()
        self.cond=threading.Condition()

    def produce(self):
        for b in range(100):
            with self.cond:
                self.data=random.randint(1,100)
                logging.info(‘produce {}‘.format(self.data))
                # self.cond.notify_all()
                self.cond.notify(2)
            self.event.wait(1)

    def consume(self):
        while True:
            with self.cond:
                self.cond.wait()
                logging.info(‘thread {} consume {}‘.format(threading.current_thread(),self.data))
            # self.event.wait(0.5)

v=Dispatcher()
# threading.Thread(target=v.produce).start()
# threading.Event().wait(3)
# threading.Thread(target=v.consume).start()
# threading.Thread(target=v.consume).start()

for b in range(5):
    threading.Thread(target=v.consume,name=‘consumer-{}‘.format(b)).start()
threading.Thread(target=v.produce).start()

 

import threading,logging,random
FORMAT=‘%(asctime)s [%(threadName)s %(thread)d] %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.CRITICAL)

class Dispatcher:
    def __init__(self):
        self.data=None
        self.event=threading.Event()
    def produce(self,total):
        for _ in range(total):
            data=random.randrange(1,100)
            logging.critical(data)
            self.data=data
            self.event.wait(1)
        self.event.set() # 结束consume的endless loop

    def consume(self):
        while not self.event.is_set():
            logging.critical(‘received {}‘.format(self.data))
            self.data=None
            self.event.wait(0.5)

v=Dispatcher()
threading.Thread(target=v.consume,name=‘consumer‘).start()
threading.Event().wait(2)
threading.Thread(target=v.produce,args=(10,),name=‘producer‘).start()

 

import threading,logging,time
FORMAT=‘%(name)s %(asctime)s %(threadName)s %(thread)d %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.INFO)

cups=[]

def worker(lock:threading.Lock,task=100):
    while True:
        if lock.acquire(blocking=False):
            count=len(cups)
            logging.info(‘number of cups: {}‘.format(count))

            if count>=task: # 每个子线程必须执行此退出条件
                lock.release()
                break

            cups.append(1)
            lock.release()
            logging.info(‘{} produce {} cup‘.format(threading.current_thread().name,len(cups)))
            time.sleep(0.5)
    logging.info(‘worker end,number of cups: {}‘.format(len(cups)))

lock=threading.Lock()
for _ in range(10):
    threading.Thread(target=worker,args=(lock,40)).start()

while True:
    if threading.active_count() == 1:
        print(threading.enumerate())
        print(len(cups))
    else:
        print(threading.enumerate())
    time.sleep(1)

 

import threading,logging,random
FORMAT=‘%(asctime)s [ %(threadName)s %(thread)d ] %(message)s‘
logging.basicConfig(format=FORMAT,level=logging.ERROR)

class Dispatcher:
    def __init__(self):
        self.data=None
        self.event=threading.Event()
        self.cond=threading.Condition()

    def produce(self,total):
        for _ in range(total):
            with self.cond:
                data=random.randint(1,100)
                logging.error(data)
                self.data=data
                self.cond.notify_all()
            self.event.wait(1)

        self.event.set()
        print(88888888888888888888888888)
        with self.cond: # notify最后一次使consumer结束循环
            self.cond.notify_all()

    def consumer(self):
        while not self.event.is_set():
            with self.cond:
                print(3333333333)
                self.cond.wait()
                print(444444444444444444444)
                logging.error(‘received {}‘.format(self.data))
                self.data=None
            self.event.wait(0.5)
            print(self.event.is_set())

v=Dispatcher()
threading.Thread(target=v.produce,args=(5 ,),name=‘producer‘).start()
threading.Event().wait(1)
threading.Thread(target=v.consumer,name=‘consumer‘).start()

 

多线程

标签：count   lazy   false   图片   inf   png   释放   cti   finish   

原文地址：https://www.cnblogs.com/dissipate/p/13875217.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！