42线程4_Semaphore_GIL

专注于为中小企业提供成都网站建设、成都网站设计服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业玉溪免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了上千余家企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。

threading.Semaphore类：...1

数据结构Queue：...5

GIL.5

threading.Semaphore类：

线程同步技术；

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

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

与linux OS的信号完全不同；

s=threading.Semaphore(value=1)，构造方法，value<0抛ValueError异常；

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())

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: 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()

print(s.__dict__)

输出：

{'_value': 6, '_cond': , 0)>}

例，解决超出预设值：

s = threading.BoundedSemaphore(3)

s.release()

输出：

Traceback (most recent call last):

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

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就抛异常，解决：用Lock或Semaphore

# 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个就满了，有三个线程A、B、C都执行了第一句，都没来得及release，这时轮到线程A release，然后轮到C release，这时一定出问题，超界了ValueError，因此有界信号量能保证，一定不能多归还；

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

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

数据结构Queue：

标准库；

提供FIFO和LIFO的Queue，优先队列；

Queue类是线程安全的，适用于多线程间安全的交换数据，内部使用了Lock和Condition；

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

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

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

GIL

gobal intepreter lock，全局解释器锁；

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

cpython中没有真正的多线程；

GIL的本质：理解cpython多线程；

cpython中：

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

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

IO密集型，使用多线程；

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

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

如果非要使用多线程且要有效率，请绕行，选择其它语言go、erlang等；

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()

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的存在；

当前文章：42线程4_Semaphore_GIL
当前地址：http://scyanting.com/article/igicog.html

42线程4_Semaphore_GIL

threading.Semaphore类：

数据结构Queue：

GIL

其他资讯