温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

42线程4_Semaphore_GIL

发布时间:2020-07-11 23:37:25 来源:网络 阅读:361 作者:chaijowin 栏目:编程语言

 

 

目录

threading.Semaphore类:... 1

数据结构Queue... 5

GIL. 5

 

 

 

threading.Semaphore类:

线程同步技术;

Lock很像,信号量内部维护一个倒计数器,每一次acquire都会减1release1),当acquire方法发现计数为0就阻塞,直至其它线程对信号量release后,计数大于0,恢复阻塞的线程;

 

倒计数器永远不会<0,是非负数,因为acquire时发现是0都会被阻塞;

linux OS的信号完全不同;

 

s=threading.Semaphore(value=1),构造方法,value<0ValueError异常;

s.acquire(blocking=True,timeout=None),获取信号量,计数器减1,获取成功返回True

s.release(),释放信号量,计数器加1

 

应用场景:

连接池;

因为资源有限,每开启一个,连接成本很高(TCP三次握手四次挥手等),所以使用连接池;

 

threading.BoundedSemaphore类:

有界的信号量,作上界控制;

不允许使用release超出初始值的范围,否则抛ValueError异常;

 

信号量和锁:

锁,只允许同一个时间一个线程独占资源,它是特殊的信号量,即信号量计数器初值为1

信号量,允许多个线程访问共享资源,但这个共享资源数量有限;

锁,可看作是特殊的信号量;

 

例:

def work(s:threading.Semaphore):

    logging.info('in sub')

    s.acquire()

    logging.info('end sub')

 

s = threading.Semaphore(3)

 

logging.info(s.acquire())

logging.info(s.acquire())

logging.info(s.acquire())

 

threading.Thread(target=work, args=(s,)).start()

 

print('~'*20)

time.sleep(2)

logging.info(s.acquire(False))

logging.info(s.acquire(timeout=3))

s.release()

print('end main')

输出:

2018-08-07-10:07:23       Thread info: 9360 MainThread True

2018-08-07-10:07:23       Thread info: 9360 MainThread True

2018-08-07-10:07:23       Thread info: 9360 MainThread True

2018-08-07-10:07:23       Thread info: 13180 Thread-1 in sub

~~~~~~~~~~~~~~~~~~~~

2018-08-07-10:07:25       Thread info: 9360 MainThread False

2018-08-07-10:07:28       Thread info: 9360 MainThread False

2018-08-07-10:07:28       Thread info: 13180 Thread-1 end sub

end main

 

例:

s = threading.Semaphore(3)

s.release()   #直接使用release改变了初始预设值

s.release()

s.release()

print(s.__dict__)

输出:

{'_value': 6, '_cond': <Condition(<unlocked _thread.lock object at 0x00000000012AB5D0>, 0)>}

 

例,解决超出预设值:

s = threading.BoundedSemaphore(3)

s.release()

输出:

Traceback (most recent call last):

  File "E:/git_practice/cmdb/example_threading2.py", line 359, in <module>

    s.release()

  File "D:\Python\Python35\lib\threading.py", line 480, in release

    raise ValueError("Semaphore released too many times")

ValueError: Semaphore released too many times

 

例:

连接池应有容量(总数),有一个工厂方法可获取连接,能够把不用的连接返回,供其它调用者使用;

 

class Conn:

    def __init__(self, name):

        self.name = name

 

class Pool:

    def __init__(self, count):

        self.count = count

        # self.pool = []

        self.pool = [self._connect('conn-{}'.format(i)) for i in range(count)]

 

    def _connect(self, conn_name):

        return Conn(conn_name)

 

    def get_conn(self):   #此方法在多线程时有安全问题;如果池中正好有一个连接,有可能多个线程判断池的长度是>0的,当一个线程拿走了连接对象,其它线程再来pop就抛异常,解决:用LockSemaphore

        # return self.pool.pop()

        if len(self.pool) > 0:

            return self.pool.pop()

 

    def return_conn(self, conn:Conn):

        self.pool.append(conn)

 

例,连接池改进:

class Conn:

    def __init__(self, name):

        self.name = name

 

class Pool:

    def __init__(self, count):

        self.count = count

        self.pool = [self._connect('conn-{}'.format(i)) for i in range(count)]

        # self.sema = threading.Semaphore(count)

        self.sema = threading.BoundedSemaphore(count)   #信号量比计算列表长度要好

 

    def _connect(self, conn_name):

        return Conn(conn_name)

 

    def get_conn(self):

        self.sema.acquire()

        data = self.pool.pop()

        return data

 

    def return_conn(self, conn:Conn):

        self.pool.append(conn)

        self.sema.release()

 

pool = Pool(3)

def work(pool:Pool):

    conn = pool.get_conn()

    logging.info(conn)

    threading.Event().wait(random.randint(1,4))

    pool.return_conn(conn)

 

for i in range(6):

    threading.Thread(target=work, args=(pool,), name='worker-{}'.format(i)).start()

输出:

2018-08-07-14:09:11       Thread info: 3264 worker-0 <__main__.Conn object at 0x00000000014E6780>

2018-08-07-14:09:11       Thread info: 12452 worker-1 <__main__.Conn object at 0x00000000014E6630>

2018-08-07-14:09:11       Thread info: 11468 worker-2 <__main__.Conn object at 0x00000000014E6748>

2018-08-07-14:09:13       Thread info: 13228 worker-3 <__main__.Conn object at 0x00000000014E6748>

2018-08-07-14:09:14       Thread info: 12692 worker-4 <__main__.Conn object at 0x00000000014E6780>

2018-08-07-14:09:15       Thread info: 6748 worker-5 <__main__.Conn object at 0x00000000014E6630>

 

注:

使用信号量解决资源有限问题;

如果池中有资源,请求者获取资源时信号量减1,拿走资源;

当请求超过资源数,请求者只能等待;

当使用者用完归还资源后信号量加1,等到线程拿到就可唤醒拿走资源;

 

以上,从程序逻辑上分析:

1、如果还没有使用信号量就release,会怎么样?

超出预设值,解决用threading.BoundedSemaphore

2、如果使用了信号量,但还没用完?

        self.pool.append(conn)

        self.sema.release()

如果一种极端情况,计数器还差1个就满了,有三个线程ABC都执行了第一句,都没来得及release,这时轮到线程A release,然后轮到C release,这时一定出问题,超界了ValueError,因此有界信号量能保证,一定不能多归还;

3、很多线程用完了信号量?

没有获得信号量的线程都阻塞,没有线程和归还的线程争抢,当append后才release,这时才能等待的线程被唤醒,才能pop,即没有获取信号量就不能pop,这是安全的;

 

 

 

数据结构Queue

标准库;

提供FIFOLIFOQueue,优先队列;

Queue类是线程安全的,适用于多线程间安全的交换数据,内部使用了LockCondition

 

魔术方法中,说实现容器的大小不准确?

如果不加锁,是不可能获得准确的大小的,因为当前线程刚读取了一个大小,还没取走,就有可能被其它线程改掉了;

 

Queue类的size虽加了锁,但依然不能保证立即getput就能成功,因为读取大小和getput方法是分开的;

 

 

 

GIL

gobal intepreter lock,全局解释器锁;

GIL保证cpyton进程中,只有一个线程执行字节码,甚至在多核cpu情况下也是如此;ruby中也有GIL;另,其它解释器没有这种情况,如pypyjython等;

 

cpython中没有真正的多线程;

GIL的本质:理解cpython多线程;

 

cpython中:

IO密集型,由于线程阻塞,就会调度其它线程;

cpu密集型,当前线程可能会连续的获得GIL,导致其它线程几乎无法使用cpu

 

IO密集型,使用多线程;

cpu密集型,使用多进程,绕开GIL

 

新版cpython正努力优化GIL问题,但不是移除;

如果非要使用多线程且要有效率,请绕行,选择其它语言goerlang等;

 

py中绝大多数内置数据结构都是原子操作,如lst.append()lst.pop()

由于GIL的存在,py的内置数据类型在多线程编程时就变为安全的了,但实际上它们本身不是线程安全的;

 

项目一开始就要固定使用某一解释器的XX版本;

 

保留GIL的原因:

guido坚持的简单哲学,对于初学者门槛低,不需要高深的系统知识也能安全、简单的使用py

若移除GIL,会降低cpython单线程的执行效率;

 

1

def calc():

    sum = 0

    for _ in range(100000000):

        sum += 1

 

start = datetime.datetime.now()

 

calc()

calc()

calc()

calc()

calc()

 

delta = (datetime.datetime.now() - start).total_seconds()

print(delta)

输出:

32.942885

 

2

def calc():

    sum = 0

    for _ in range(100000000):

        sum += 1

 

start = datetime.datetime.now()

 

lst = []

for _ in range(5):   #CPU密集型,用多进程解决

    t = threading.Thread(target=calc)

    t.start()

    # t.join()   #t.join()不能放在此处,若放在此处是串行(第1个线程执行完才能循环执行到下一个线程),不是并行

    lst.append(t)

for t in lst:

    t.join()

 

delta = (datetime.datetime.now() - start).total_seconds()

print(delta)

输出:

32.487859

 

对比例1和例2

从结果上看,单线程和多线程一样,cpython的多线程没有优势和一个线程执行时间相当,因为GIL的存在;

 

 


向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI