MindMap
高清原图: 在此
Thread Objects
GIL: global interpreter lock
The mechanism used by the CPython interpreter to assure that only one thread executes Python bytecode at a time.class threading.Thread(.., target=None, .., args=(), kwargs={}, .., ..)
start() 函数调用 run(), run() 调用 target(*args, **kwargs).Thread.__init__()
If the subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.
class Mythread(threading.Thread):
def __init__(self):
super(Mythread, self).__init__()
...
def run(self):
...
Lock/RLock Objects
class threading.RLock
acquire(blocking=True, timeout=-1)
When invoked without arguments: if this thread already owns the lock, increment the recursion level by one, and return immediately. Otherwise, if another thread owns the lock, block until the lock is unlocked. Once the lock is unlocked (not owned by any thread), then grab ownership, set the recursion level to one, and return. If more than one thread is blocked waiting until the lock is unlocked, only one at a time will be able to grab ownership of the lock. There is no return value in this case.
...release()
Release a lock, decrementing the recursion level. If after the decrement it is zero, reset the lock to unlocked (not owned by any thread), and if any other threads are blocked waiting for the lock to become unlocked, allow exactly one of them to proceed. If after the decrement the recursion level is still nonzero, the lock remains locked and owned by the calling thread.
官方文档结合代码一起看，更清楚:
- in Lib/threading.py
class _RLock:
...
def acquire(self, blocking=True, timeout=-1):
me = get_ident()
if self._owner == me:
self._count += 1
return 1
rc = self._block.acquire(blocking, timeout)
if rc:
self._owner = me
self._count = 1
return rc
...
def release(self):
if self._owner != get_ident():
raise RuntimeError("cannot release un-acquired lock")
self._count = count = self._count - 1
if not count:
self._owner = None
self._block.release()
...
...
Condition Objects
苍松 同学说: 可以认为，除了Lock带有的锁定池外，Condition还包含一个等待池，池中的线程处于等待阻塞状态，直到另一个线程调用notify()/notifyAll()通知；得到通知后线程进入锁定池等待锁定。
in Lib/threading.py
class Condition:
def __init__(self, lock=None):
if lock is None: # If the lock argument is None, a new RLock object is created and used as the underlying lock.
lock = RLock()
self._lock = lock
...
self._waiters = _deque() # 等待池
...
def wait(self, timeout=None):
if not self._is_owned(): # If the calling thread has not acquired the lock when this method is called, a RuntimeError is raised.
raise RuntimeError("cannot wait on un-acquired lock")
waiter = _allocate_lock()
waiter.acquire()
self._waiters.append(waiter) # 将它放到等待池中
saved_state = self._release_save() # 释放锁
gotit = False
try: # restore state no matter what (e.g., KeyboardInterrupt)
if timeout is None:
waiter.acquire() # 获取锁
gotit = True
else:
if timeout > 0:
gotit = waiter.acquire(True, timeout)
else:
gotit = waiter.acquire(False)
return gotit
finally:
self._acquire_restore(saved_state)
if not gotit:
try:
self._waiters.remove(waiter)
except ValueError:
pass
def wait_for(self, predicate, timeout=None):
endtime = None
waittime = timeout
result = predicate()
while not result:
if waittime is not None:
if endtime is None:
endtime = _time() + waittime
else:
waittime = endtime - _time()
if waittime <= 0:
break
self.wait(waittime)
result = predicate()
return result
def notify(self, n=1):
if not self._is_owned():
raise RuntimeError("cannot notify on un-acquired lock")
all_waiters = self._waiters
waiters_to_notify = _deque(_islice(all_waiters, n))
if not waiters_to_notify:
return
for waiter in waiters_to_notify:
waiter.release() # 唤醒等待池的线程，并把它放到锁定池中
try:
all_waiters.remove(waiter) # 移除已被唤醒了的线程
except ValueError:
pass
def notify_all(self):
self.notify(len(self._waiters)) # 唤醒所有在等待池中的线程
Semaphore Objects
A semaphore manages an internal counter which is decremented by each acquire() call and incremented by each release() call. The counter can never go below zero; when acquire() finds that it is zero, it blocks, waiting until some other thread calls release().
- in Lib/threading.py
class Semaphore:
def __init__(self, value=1):
if value < 0:
raise ValueError("semaphore initial value must be >= 0")
self._cond = Condition(Lock())
self._value = value
def acquire(self, blocking=True, timeout=None):
if not blocking and timeout is not None:
raise ValueError("can't specify timeout for non-blocking acquire")
rc = False
endtime = None
with self._cond:
while self._value == 0:
if not blocking:
break
if timeout is not None:
if endtime is None:
endtime = _time() + timeout
else:
timeout = endtime - _time()
if timeout <= 0:
break
self._cond.wait(timeout)
else:
self._value -= 1
rc = True
return rc
__enter__ = acquire
def release(self):
with self._cond:
self._value += 1
self._cond.notify()
def __exit__(self, t, v, tb):
self.release()
class BoundedSemaphore(Semaphore):
def __init__(self, value=1):
Semaphore.__init__(self, value)
self._initial_value = value
def release(self):
with self._cond:
if self._value >= self._initial_value:
raise ValueError("Semaphore released too many times")
self._value += 1
self._cond.notify()
Event Objects
one thread signals an event and other threads wait for it.
苍松 同学说:
Event其实就是一个简化版的 Condition。Event没有锁，无法使线程进入同步阻塞状态。
－In Lib/threading.py
class Event:
def __init__(self):
self._cond = Condition(Lock())
self._flag = False
...
def is_set(self):
return self._flag
isSet = is_set
def set(self):
with self._cond:
self._flag = True
self._cond.notify_all() # 唤醒所有在等待阻塞中的线程
def clear(self):
with self._cond:
self._flag = False
def wait(self, timeout=None):
with self._cond:
signaled = self._flag
if not signaled:
signaled = self._cond.wait(timeout)
return signaled
Timer Objects
This class represents an action that should be run only after a certain amount of time has passed — a timer.
- in Lib/threading.py
class Timer(Thread):
def __init__(self, interval, function, args=None, kwargs=None):
Thread.__init__(self)
self.interval = interval
self.function = function
self.args = args if args is not None else []
self.kwargs = kwargs if kwargs is not None else {}
self.finished = Event()
def cancel(self):
"""Stop the timer if it hasn't finished yet."""
self.finished.set() # set flag=True, and then the return of is_set() is True, so it would not execute the function.
def run(self):
self.finished.wait(self.interval) # 在此等待阻塞中可能被取消
if not self.finished.is_set(): # 若上一步出来，便表示没被取消，flag = False, 运行此function.
self.function(*self.args, **self.kwargs)
self.finished.set()
Barrier Obects
This class provides a simple synchronization primitive for use by a fixed number of threads that need to wait for each other.
来自 Mayank Kumar 同学的例子
import threading
barrier = threading.Barrier(3)
class thread(threading.Thread):
def __init__(self, thread_ID):
threading.Thread.__init__(self)
self.thread_ID = thread_ID
def run(self):
print(str(self.thread_ID) + "\n")
print("Parties = " + str(barrier.parties) + "\n")
print("n_waiting = " + str(barrier.n_waiting) + "\n")
barrier.wait()
thread1 = thread(100)
thread2 = thread(101)
thread1.start()
thread2.start()
barrier.wait()
print(str(barrier.broken) + "\n")
barrier.reset()
print("n_waiting after reset = " + str(barrier.n_waiting))
barrier.abort()
print("End")
运行结果:
100
Parties = 3
101
n_waiting = 0
Parties = 3
n_waiting = 1
False
n_waiting after reset = 0
End
with statement
All of the objects provided by this module that have acquire() and release() methods can be used as context managers for a with statement. The acquire() method will be called when the block is entered, and release() will be called when the block is exited.
with some_lock:
# do something...
is equivalent to
some_lock.acquire()
try:
# do something...
finally:
some_lock.release()
read more
Python ducumentation: threading
PyMOTW-3: threading — Manage Concurrent Operations Within a Process
苍松 同学的 python--threading多线程总结
tab609 同学的 Python多线程详解
