Python 线程|进程

Python   线程

Threading是用于提供线程相关的操作，线程是应用程序中工作的最小单元。线程与进程的关系下图所示：

　　子线程是由主线程产生的，但两者并没有关联。

利用threading创建线程：

'''利用threading包创建'''
import threading
import time

def run(n):
    time.sleep(2)
    print("task:",n)

'''串行：一个运行完后，再运行另外一个'''
run("t1")  #并不是线程，只是调用方法传参数
run("t2")

'''并发性'''
t1 = threading.Thread(target=run,args=("T1",))  #t1是线程，args为元组
t2 = threading.Thread(target=run,args=("T2",))
t1.start()  #并发性地工作
t2.start()


'''运行结果'''
task: t1         #t1运行后会间隔两秒，然后运行t2
task: t2

task: T2         #T1,T2同时运行
task: T1

上述创建了两个线程t1和t2,然后控制器就交给了CPU，CPU根据指定算法进行调度，分片执行指令。

更多方法：

• start    线程准备就绪，等待CPU调度；启动线程的活动，每个线程对象最多只能调用一次。
• join     逐个执行每个线程，执行完毕后继续往下执行，该方法使得多线程变得无意义。
• run      表示线程活动的方法。可以在子类中重写此方法。标准run()方法调用传递给对象构造函数的可调用对象作为目标参数(如果有的话)，分别使用args和kwargs参数中的顺序参数和关键字参数。线程被cpu调度后自动执行线程对象的run方法
• get_ident()    获得线程地址
• setName    为线程设置名称
• getName    获取线程名称
• daemon     一个布尔值，指示此线程是否为守护线程。这必须在调用start()之前设置，否则会引发运行时错误。它的初始值继承自创建线程;主线程不是守护进程线程，因此在主线程中创建的所有线程默认为守护进程= False。当没有存活的非守护进程线程时，整个Python程序退出。
• setDaemon   设置为后台线程或前台线程（默认）如果是后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止；如果是前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止

#子线程是由主线程产生的，但两者并没有关联
import threading
import time

def run(n):
    print("task:",n)
    time.sleep(0.1)
    print("taskdone:",n)

Start_time = time.time()
for i in range(50):     #共有51个线程，代码本身是一个主线程
    t = threading.Thread(target=run,args=("t--%s" % i,))
    t.start()
    t.join()            #join使得主线程与子线程成串行运行

print(time.time()-Start_time)  #print为创建子线程所产生的时间，而非运行时间

import threading
import time

class My_Thread(threading.Thread):
    def __init__(self,n):
        super(My_Thread,self).__init__()
        self.n = n

    def  run(self):
        print("task:",self.n)
        time.sleep(0.1)
t_obj=[]
start_time = time.time()
for i in range(50):     #共有51个线程，代码本身是一个主线程
    t = My_Thread("t--%s" % i)
    t.setDaemon(True)   #监听端口，当主程序执行完毕，将不会执行其他线程（前提是去掉join方法）
    t.start()
    t_obj.append(t)
print(time.time()-start_time)


'''运行结果'''
task: t--0
task: t--1
task: t--2
task: t--3
task: t--4
task: t--5
task: t--6
task: t--7
task: t--8
task: t--9
task: t--10
task: t--11
task: t--12
task: t--13
task: t--14
task: t--15
task: t--16
task: t--17
task: t--18
task: t--19
task: t--20
task: t--21
task: t--22
task: t--23
task: t--24
task: t--25
task: t--26
task: t--27
task: t--28
task: t--29
task: t--30
task: t--31
task: t--32
task: t--33
task: t--34
task: t--35
task: t--36
task: t--37
task: t--38
task: t--39
task: t--40
task: t--41
task: t--42
task: t--43
task: t--44
task: t--45
task: t--46
task: t--47
task: t--48
task: t--49
0.01196908950805664

线程锁（Lock）：

    def acquire(self, blocking=True, timeout=None):
        """Acquire a semaphore, decrementing the internal counter by one.
        When invoked without arguments: if the internal counter is larger than
        zero on entry, decrement it by one and return immediately. If it is zero
        on entry, block, waiting until some other thread has called release() to
        make it larger than zero. This is done with proper interlocking so that
        if multiple acquire() calls are blocked, release() will wake exactly one
        of them up. The implementation may pick one at random, so the order in
        which blocked threads are awakened should not be relied on. There is no
        return value in this case.
        When invoked with blocking set to true, do the same thing as when called
        without arguments, and return true.
        When invoked with blocking set to false, do not block. If a call without
        an argument would block, return false immediately; otherwise, do the
        same thing as when called without arguments, and return true.
        When invoked with a timeout other than None, it will block for at
        most timeout seconds.  If acquire does not complete successfully in
        that interval, return false.  Return true otherwise.
        """
        #获得一个信号量，将内部计数器减1。在没有参数的情况下调用时:如果内部计数器在入口时
        # 大于0，则将其递减1并立即返回。如果进入时为零，阻塞，等待其他线程调用release()
        # 使其大于零。这是通过适当的联锁完成的，这样，如果多个acquire()调用被阻塞，
        # release()就会唤醒其中一个调用。实现可以随机选择一个线程，因此不应该依赖于被阻塞
        # 线程被唤醒的顺序。在本例中没有返回值。当阻塞集调用为true时，执行与没有参数调用
        # 时相同的操作，并返回true。当阻塞设置为false时，不要阻塞。如果一个没有参数的
        # 调用将阻塞，立即返回false;否则，执行与没有参数调用时相同的操作，并返回true。
        # 当使用除None以外的超时调用时，它最多将阻塞超时秒。如果在那段时间里收购没有成功
        # 完成，还假。否则返回true。
        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):
        """Release a semaphore, incrementing the internal counter by one.
        When the counter is zero on entry and another thread is waiting for it
        to become larger than zero again, wake up that thread.
        """
        #释放信号量，增加一个内部计数器。当进入时计数器为零，而另一个线程正在等待计数器
        # 再次大于零时，唤醒该线程。
        with self._cond:
            self._value += 1
            self._cond.notify()

    def __exit__(self, t, v, tb):
        self.release()

import threading
import time

lock = threading.Lock()   #线程锁

def run(n):
    lock.acquire()             #锁定
    global num
    num+=1
    lock.release()              #释放锁
    time.sleep(1)

t_obj = []
num = 0
for i in range(50):
    t = threading.Thread(target=run,args=("t--%s" % i,))
    t.start()
    t_obj.append(t)

for i in t_obj:
    i.join()

print("num:",num)


'''运行结果'''
num: 50

'''可用来做测试'''

if __name__ == "__main__"  

#表示函数的开始位置，判断自主运行与否

　　线程池（信号量（semaphore））：

　　信号量管理一个计数器，该计数器表示release()调用的数量减去acquire()调用的数量，再加上一个初始值。acquire()方法如果有必要会阻塞，直到它可以返回而不会使计数器变为负数为止。如果未指定，值默认为1。

'''信号量'''
import threading
import time

def run(n):
    Semaphore.acquire()
    print("task:",n)
    time.sleep(1)
    Semaphore.release()

if __name__ == "__main__":
    Semaphore = threading.BoundedSemaphore(5)
    #每五个子进程运行一次，间隔一秒后，再运行下五个
    for i in range(20):
        t = threading.Thread(target=run,args=(i,))

        t.start()
    while threading.active_count()!=1:
        pass
    else:
        print("--all threading has done")

"""Thread module emulating a subset of Java's threading model."""
#线程模块模拟Java线程模型的一个子集。
import os as _os
import sys as _sys
import _thread

from time import monotonic as _time
from traceback import format_exc as _format_exc
from _weakrefset import WeakSet
from itertools import islice as _islice, count as _count
try:
    from _collections import deque as _deque
except ImportError:
    from collections import deque as _deque

# Note regarding PEP 8 compliant names
#  This threading model was originally inspired by Java, and inherited
# the convention of camelCase function and method names from that
# language. Those original names are not in any imminent danger of
# being deprecated (even for Py3k),so this module provides them as an
# alias for the PEP 8 compliant names
# Note that using the new PEP 8 compliant names facilitates substitution
# with the multiprocessing module, which doesn't provide the old
# Java inspired names.

__all__ = ['get_ident', 'active_count', 'Condition', 'current_thread',
           'enumerate', 'main_thread', 'TIMEOUT_MAX',
           'Event', 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Thread',
           'Barrier', 'BrokenBarrierError', 'Timer', 'ThreadError',
           'setprofile', 'settrace', 'local', 'stack_size']

# Rename some stuff so "from threading import *" is safe
_start_new_thread = _thread.start_new_thread
_allocate_lock = _thread.allocate_lock
_set_sentinel = _thread._set_sentinel
get_ident = _thread.get_ident
ThreadError = _thread.error
try:
    _CRLock = _thread.RLock
except AttributeError:
    _CRLock = None
TIMEOUT_MAX = _thread.TIMEOUT_MAX
del _thread


# Support for profile and trace hooks
#支持配置文件和跟踪挂钩

_profile_hook = None
_trace_hook = None

def setprofile(func):
    """Set a profile function for all threads started from the threading module.
    The func will be passed to sys.setprofile() for each thread, before its
    run() method is called.
    """
    #为从线程模块启动的所有线程设置一个配置文件函数。在调用其run()方法之前，
    # func将被传递给每个线程的sys.setprofile()。

    global _profile_hook
    _profile_hook = func

def settrace(func):
    """Set a trace function for all threads started from the threading module.
    The func will be passed to sys.settrace() for each thread, before its run()
    method is called.
    """
    #为从线程模块启动的所有线程设置跟踪函数。在调用其run()方法之前，
    # func将被传递给每个线程的sys.settrace()。

    global _trace_hook
    _trace_hook = func

# Synchronization classes
# 同步类

Lock = _allocate_lock

def RLock(*args, **kwargs):
    """Factory function that returns a new reentrant lock.
    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it again
    without blocking; the thread must release it once for each time it has
    acquired it.
    """
    #返回一个新的可重入锁的工厂函数。可重入锁必须由获得它的线程释放。
    # 一旦一个线程获得了可重入锁，该线程可以在不阻塞的情况下再次获得该锁;
    # 线程每次获得它时都必须释放它一次。

    if _CRLock is None:
        return _PyRLock(*args, **kwargs)
    return _CRLock(*args, **kwargs)

class _RLock:
    """This class implements reentrant lock objects.
    A reentrant lock must be released by the thread that acquired it. Once a
    thread has acquired a reentrant lock, the same thread may acquire it
    again without blocking; the thread must release it once for each time it
    has acquired it.
    """
    #该类实现可重入锁对象。可重入锁必须由获得它的线程释放。一旦一个线程获得了可重入锁，
    # 该线程可以在不阻塞的情况下再次获得该锁;线程每次获得它时都必须释放它一次。

    def __init__(self):
        self._block = _allocate_lock()
        self._owner = None
        self._count = 0

    def __repr__(self):
        owner = self._owner
        try:
            owner = _active[owner].name
        except KeyError:
            pass
        return "<%s %s.%s object owner=%r count=%d at %s>" % (
            "locked" if self._block.locked() else "unlocked",
            self.__class__.__module__,
            self.__class__.__qualname__,
            owner,
            self._count,
            hex(id(self))
        )

    def acquire(self, blocking=True, timeout=-1):
        """Acquire a lock, blocking or non-blocking.
        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.
        When invoked with the blocking argument set to true, do the same thing
        as when called without arguments, and return true.
        When invoked with the blocking argument set to false, do not block. If a
        call without an argument would block, return false immediately;
        otherwise, do the same thing as when called without arguments, and
        return true.
        When invoked with the floating-point timeout argument set to a positive
        value, block for at most the number of seconds specified by timeout
        and as long as the lock cannot be acquired.  Return true if the lock has
        been acquired, false if the timeout has elapsed.
        """
        #获得一个锁，阻塞或非阻塞。在没有参数的情况下调用时:如果这个线程已经拥有锁，
        # 那么将递归级别增加1，并立即返回。否则，如果另一个线程拥有锁，
        # 则阻塞直到锁被解锁。一旦锁被解锁(不属于任何线程)，然后获取所有权，
        # 将递归级别设置为1，然后返回。如果有多个线程被阻塞，等待锁被解锁，
        # 每次只有一个线程能够获取锁的所有权。在本例中没有返回值。当阻塞参数设置
        # 为true时，执行与没有参数时相同的操作，并返回true。当阻塞参数设置为false时，
        # 不要阻塞。如果一个没有参数的调用将阻塞，立即返回false;否则，执行与没有
        # 参数调用时相同的操作，并返回true。当将浮点超时参数设置为正值时，如果获得
        # 了锁，则最多阻塞超时指定的秒数，如果超时已过，则返回true;如果超时已过，则返回false。

        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

    __enter__ = acquire

    def release(self):
        """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.
        Only call this method when the calling thread owns the lock. A
        RuntimeError is raised if this method is called when the lock is
        unlocked.
        There is no return value.
        """
        #释放锁，降低递归级别。如果减量后为零，则将锁重置为解锁(不属于任何线程)，
        # 如果任何其他线程被阻塞，等待锁解锁，则只允许其中一个线程继续执行。如果在递减
        # 之后递归级别仍然是非零，则锁仍然被锁定，并且由调用线程拥有。只有当调用线程拥有
        # 锁时才调用此方法。如果在解锁锁时调用此方法，将引发运行时错误。没有返回值。

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

    def __exit__(self, t, v, tb):
        self.release()

    # Internal methods used by condition variables
    #条件变量使用的内部方法

    def _acquire_restore(self, state):
        self._block.acquire()
        self._count, self._owner = state

    def _release_save(self):
        if self._count == 0:
            raise RuntimeError("cannot release un-acquired lock")
        count = self._count
        self._count = 0
        owner = self._owner
        self._owner = None
        self._block.release()
        return (count, owner)

    def _is_owned(self):
        return self._owner == get_ident()

_PyRLock = _RLock


class Condition:
    """Class that implements a condition variable.
    A condition variable allows one or more threads to wait until they are
    notified by another thread.
    If the lock argument is given and not None, it must be a Lock or RLock
    object, and it is used as the underlying lock. Otherwise, a new RLock object
    is created and used as the underlying lock.
    """
    #实现条件变量的类。条件变量允许一个或多个线程等待，直到另一个线程通知它们。
    # 如果锁参数是给定的而不是空的，那么它必须是一个锁或RLock对象，并且它被用作底层锁。
    # 否则，将创建一个新的RLock对象并将其用作底层锁。

    def __init__(self, lock=None):
        if lock is None:
            lock = RLock()
        self._lock = lock
        # Export the lock's acquire() and release() methods
        #导出锁的acquire()和release()方法
        self.acquire = lock.acquire
        self.release = lock.release
        # If the lock defines _release_save() and/or _acquire_restore(),
        # these override the default implementations (which just call
        # release() and acquire() on the lock).  Ditto for _is_owned().
        #如果锁定义了_release_save()和/或_acquire_restore()，就会覆盖默认的实现
        # (它只调用release()和acquire()对锁进行访问)。_is_owned同上()。
        try:
            self._release_save = lock._release_save
        except AttributeError:
            pass
        try:
            self._acquire_restore = lock._acquire_restore
        except AttributeError:
            pass
        try:
            self._is_owned = lock._is_owned
        except AttributeError:
            pass
        self._waiters = _deque()

    def __enter__(self):
        return self._lock.__enter__()

    def __exit__(self, *args):
        return self._lock.__exit__(*args)

    def __repr__(self):
        return "<Condition(%s, %d)>" % (self._lock, len(self._waiters))

    def _release_save(self):
        self._lock.release()           # No state to save 没有状态保存

    def _acquire_restore(self, x):
        self._lock.acquire()           # Ignore saved state 忽略保存的状态

    def _is_owned(self):
        # Return True if lock is owned by current_thread.
        #如果锁属于current_thread，则返回True。
        # This method is called only if _lock doesn't have _is_owned().
        #只有当_lock没有_is_owned()时才调用该方法。
        if self._lock.acquire(0):
            self._lock.release()
            return False
        else:
            return True

    def wait(self, timeout=None):
        """Wait until notified or until a timeout occurs.
        If the calling thread has not acquired the lock when this method is
        called, a RuntimeError is raised.
        This method releases the underlying lock, and then blocks until it is
        awakened by a notify() or notify_all() call for the same condition
        variable in another thread, or until the optional timeout occurs. Once
        awakened or timed out, it re-acquires the lock and returns.
        When the timeout argument is present and not None, it should be a
        floating point number specifying a timeout for the operation in seconds
        (or fractions thereof).
        When the underlying lock is an RLock, it is not released using its
        release() method, since this may not actually unlock the lock when it
        was acquired multiple times recursively. Instead, an internal interface
        of the RLock class is used, which really unlocks it even when it has
        been recursively acquired several times. Another internal interface is
        then used to restore the recursion level when the lock is reacquired.
        """
        #等待直到通知或超时发生。如果调用该方法时调用的线程没有获得锁，则会引发运行时错误。
        # 该方法释放底层锁，然后阻塞，直到它被另一个线程中的notify()或notify_all()调用
        # 唤醒，或者直到出现可选超时为止。一旦被唤醒或超时，它会重新获得锁并返回。
        # 当出现timeout参数而不是None时，它应该是一个浮点数，以秒(或几分之一)为单位指定
        # 操作的超时。当底层锁是RLock时，不会使用其release()方法释放它，因为当递归地多次
        # 获取锁时，这可能不会真正解锁它。相反，使用了RLock类的内部接口，即使递归地获得了
        # 多次，它也会真正地解锁它。然后使用另一个内部接口在重新获得锁时恢复递归级别。

        if not self._is_owned():
            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):
        """Wait until a condition evaluates to True.
        predicate should be a callable which result will be interpreted as a
        boolean value.  A timeout may be provided giving the maximum time to
        wait.
        """
        #等待，直到条件的值为True。谓词应该是可调用的，其结果将被解释为布尔值。
        # 可能会提供一个超时，以提供最长的等待时间。

        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):
        """Wake up one or more threads waiting on this condition, if any.
        If the calling thread has not acquired the lock when this method is
        called, a RuntimeError is raised.
        This method wakes up at most n of the threads waiting for the condition
        variable; it is a no-op if no threads are waiting.
        """
        #唤醒在此条件下等待的一个或多个线程(如果有的话)。如果调用该方法时调用的线程没有获得锁，
        # 则会引发运行时错误。该方法最多唤醒n个等待条件变量的线程;如果没有线程在等待，那么
        # 这是一个no-op。
        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):
        """Wake up all threads waiting on this condition.
        If the calling thread has not acquired the lock when this method
        is called, a RuntimeError is raised.
        """
        #唤醒在此条件下等待的所有线程。如果调用该方法时调用的线程没有获得锁，
        # 则会引发运行时错误。
        self.notify(len(self._waiters))

    notifyAll = notify_all


class Semaphore:
    """This class implements semaphore objects.
    Semaphores manage a counter representing the number of release() calls minus
    the number of acquire() calls, plus an initial value. The acquire() method
    blocks if necessary until it can return without making the counter
    negative. If not given, value defaults to 1.
    """
    #这个类实现信号量对象。信号量管理一个计数器，该计数器表示release()调用的数量减去
    # acquire()调用的数量，再加上一个初始值。acquire()方法如果有必要会阻塞，直到它可以
    # 返回而不会使计数器变为负数为止。如果未指定，值默认为1。

    # After Tim Peters' semaphore class, but not quite the same (no maximum)
    #在Tim Peters的信号量类之后，但不完全相同(没有最大值)

    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):
        """Acquire a semaphore, decrementing the internal counter by one.
        When invoked without arguments: if the internal counter is larger than
        zero on entry, decrement it by one and return immediately. If it is zero
        on entry, block, waiting until some other thread has called release() to
        make it larger than zero. This is done with proper interlocking so that
        if multiple acquire() calls are blocked, release() will wake exactly one
        of them up. The implementation may pick one at random, so the order in
        which blocked threads are awakened should not be relied on. There is no
        return value in this case.
        When invoked with blocking set to true, do the same thing as when called
        without arguments, and return true.
        When invoked with blocking set to false, do not block. If a call without
        an argument would block, return false immediately; otherwise, do the
        same thing as when called without arguments, and return true.
        When invoked with a timeout other than None, it will block for at
        most timeout seconds.  If acquire does not complete successfully in
        that interval, return false.  Return true otherwise.
        """
        #获得一个信号量，将内部计数器减1。在没有参数的情况下调用时:如果内部计数器在入口时
        # 大于0，则将其递减1并立即返回。如果进入时为零，阻塞，等待其他线程调用release()
        # 使其大于零。这是通过适当的联锁完成的，这样，如果多个acquire()调用被阻塞，
        # release()就会唤醒其中一个调用。实现可以随机选择一个线程，因此不应该依赖于被阻塞
        # 线程被唤醒的顺序。在本例中没有返回值。当阻塞集调用为true时，执行与没有参数调用
        # 时相同的操作，并返回true。当阻塞设置为false时，不要阻塞。如果一个没有参数的
        # 调用将阻塞，立即返回false;否则，执行与没有参数调用时相同的操作，并返回true。
        # 当使用除None以外的超时调用时，它最多将阻塞超时秒。如果在那段时间里收购没有成功
        # 完成，还假。否则返回true。
        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):
        """Release a semaphore, incrementing the internal counter by one.
        When the counter is zero on entry and another thread is waiting for it
        to become larger than zero again, wake up that thread.
        """
        #释放信号量，增加一个内部计数器。当进入时计数器为零，而另一个线程正在等待计数器
        # 再次大于零时，唤醒该线程。
        with self._cond:
            self._value += 1
            self._cond.notify()

    def __exit__(self, t, v, tb):
        self.release()


class BoundedSemaphore(Semaphore):
    """Implements a bounded semaphore.
    A bounded semaphore checks to make sure its current value doesn't exceed its
    initial value. If it does, ValueError is raised. In most situations
    semaphores are used to guard resources with limited capacity.
    If the semaphore is released too many times it's a sign of a bug. If not
    given, value defaults to 1.
    Like regular semaphores, bounded semaphores manage a counter representing
    the number of release() calls minus the number of acquire() calls, plus an
    initial value. The acquire() method blocks if necessary until it can return
    without making the counter negative. If not given, value defaults to 1.
    """
    #实现有界信号量。有界信号量检查其当前值是否不超过初始值。如果是，则会引发ValueError。
    # 在大多数情况下，信号量被用来保护有限容量的资源。如果信号量被释放了太多次，这是错误
    # 的信号。如果未指定，值默认为1。与常规信号量一样，有界信号量管理一个计数器，
    # 表示release()调用的数量减去acquire()调用的数量，再加上一个初始值。acquire()方法
    # 如果有必要会阻塞，直到它可以返回而不会使计数器变为负数为止。如果未指定，值默认为1。

    def __init__(self, value=1):
        Semaphore.__init__(self, value)
        self._initial_value = value

    def release(self):
        """Release a semaphore, incrementing the internal counter by one.

        When the counter is zero on entry and another thread is waiting for it
        to become larger than zero again, wake up that thread.

        If the number of releases exceeds the number of acquires,
        raise a ValueError.
        """
        #释放信号量，增加一个内部计数器。当进入时计数器为0，而另一个线程正在等待i再次
        # 大于0时，唤醒那个线程。如果发布的数量超过了获得的数量，则引发一个ValueError。
        with self._cond:
            if self._value >= self._initial_value:
                raise ValueError("Semaphore released too many times")
            self._value += 1
            self._cond.notify()


class Event:
    """Class implementing event objects.

    Events manage a flag that can be set to true with the set() method and reset
    to false with the clear() method. The wait() method blocks until the flag is
    true.  The flag is initially false.
    """
    #类实现事件对象。事件管理的标志可以用set()方法设置为true，用clear()方法重置为false。
    # wait()方法将阻塞，直到标记为true。标志最初是假的。

    # After Tim Peters' event class (without is_posted())
    #在Tim Peters的事件类之后(没有is_post ())

    def __init__(self):
        self._cond = Condition(Lock())
        self._flag = False

    def _reset_internal_locks(self):
        # private!  called by Thread._reset_internal_locks by _after_fork()
        #私人!调用线程._reset_internal_locks _after_fork()
        self._cond.__init__(Lock())

    def is_set(self):
        """Return true if and only if the internal flag is true."""
        #当且仅当内部标志为true时返回true。
        return self._flag

    isSet = is_set

    def set(self):
        """Set the internal flag to true.
        All threads waiting for it to become true are awakened. Threads
        that call wait() once the flag is true will not block at all.
        """
        #将内部标志设置为true。等待它成真的所有线程都被唤醒。一旦标志为true，
        # 调用wait()的线程将不会阻塞。
        with self._cond:
            self._flag = True
            self._cond.notify_all()

    def clear(self):
        """Reset the internal flag to false.
        Subsequently, threads calling wait() will block until set() is called to
        set the internal flag to true again.
        """
        #将内部标志重置为false。随后，调用wait()的线程将阻塞，直到调用set()将内部标志再次设置为true。
        with self._cond:
            self._flag = False

    def wait(self, timeout=None):
        """Block until the internal flag is true.
        If the internal flag is true on entry, return immediately. Otherwise,
        block until another thread calls set() to set the flag to true, or until
        the optional timeout occurs.
        When the timeout argument is present and not None, it should be a
        floating point number specifying a timeout for the operation in seconds
        (or fractions thereof).
        This method returns the internal flag on exit, so it will always return
        True except if a timeout is given and the operation times out.
        """
        #阻塞，直到内部标志为true。如果进入时内部标志为true，则立即返回。否则，阻塞直到
        # 另一个线程调用set()将标志设置为true，或者直到出现可选超时。当出现timeout参数
        # 而不是None时，它应该是一个浮点数，以秒(或几分之一)为单位指定操作的超时。这个
        # 方法在退出时返回内部标志，因此它总是返回True，除非超时和操作超时。
        with self._cond:
            signaled = self._flag
            if not signaled:
                signaled = self._cond.wait(timeout)
            return signaled


# A barrier class.  Inspired in part by the pthread_barrier_* api and
# the CyclicBarrier class from Java.  See
'''一个障碍类。部分灵感来自于pthread_barrier_* api和来自Java的循环屏障类。看到'''
# http://sourceware.org/pthreads-win32/manual/pthread_barrier_init.html and
# http://java.sun.com/j2se/1.5.0/docs/api/java/util/concurrent/
#        CyclicBarrier.html
# for information.     ##获取信息
# We maintain two main states, 'filling' and 'draining' enabling the barrier
# to be cyclic.  Threads are not allowed into it until it has fully drained
# since the previous cycle.  In addition, a 'resetting' state exists which is
# similar to 'draining' except that threads leave with a BrokenBarrierError,
# and a 'broken' state in which all threads get the exception.
'''我们维持两种主要状态，“填充”和“排水”，使屏障是循环的。线程不允许进入它，直到它从
上一个循环中完全耗尽为止。此外，存在一种“重置”状态，类似于“耗尽”状态，只是线程留下了
一个故障的barriererror错误，以及所有线程都得到异常的“中断”状态。'''
class Barrier:
    """Implements a Barrier.
    Useful for synchronizing a fixed number of threads at known synchronization
    points.  Threads block on 'wait()' and are simultaneously once they have all
    made that call.
    """
    #实现了一个障碍。用于在已知同步点同步固定数量的线程。线程阻塞在'wait()'上，
    # 并且一旦它们都进行了该调用，就会同时阻塞。

    def __init__(self, parties, action=None, timeout=None):
        """Create a barrier, initialised to 'parties' threads.
        'action' is a callable which, when supplied, will be called by one of
        the threads after they have all entered the barrier and just prior to
        releasing them all. If a 'timeout' is provided, it is uses as the
        default for all subsequent 'wait()' calls.
        """
        #创建一个障碍，初始化为“party”线程。“action”是一个可调用的线程，当它被提供时，
        # 它将被其中一个线程在它们全部进入壁垒并释放它们之前调用。如果提供了'timeout'，
        # 那么它将用作所有后续'wait()'调用的默认值。
        self._cond = Condition(Lock())
        self._action = action
        self._timeout = timeout
        self._parties = parties
        self._state = 0 #0 filling, 1, draining, -1 resetting, -2 broken
        self._count = 0

    def wait(self, timeout=None):
        """Wait for the barrier.
        When the specified number of threads have started waiting, they are all
        simultaneously awoken. If an 'action' was provided for the barrier, one
        of the threads will have executed that callback prior to returning.
        Returns an individual index number from 0 to 'parties-1'.
        """
        #等待障碍。当指定数量的线程开始等待时，它们都同时被唤醒。如果为barrier提供了一个
        # “操作”，其中一个线程将在返回之前执行该回调。返回从0到“parties-1”的单个索引号。

        if timeout is None:
            timeout = self._timeout
        with self._cond:
            self._enter() # Block while the barrier drains. 隔离墙排水时要进行隔离。
            index = self._count
            self._count += 1
            try:
                if index + 1 == self._parties:
                    # We release the barrier
                    self._release()
                else:
                    # We wait until someone releases us
                    self._wait(timeout)
                return index
            finally:
                self._count -= 1
                # Wake up any threads waiting for barrier to drain.
                #唤醒任何等待屏障耗尽的线程。
                self._exit()

    # Block until the barrier is ready for us, or raise an exception
    # if it is broken.
    #阻止，直到障碍为我们准备好，或提出一个例外，如果它被打破。
    def _enter(self):
        while self._state in (-1, 1):
            # It is draining or resetting, wait until done正在排水或重置，等待完成
            self._cond.wait()
        #see if the barrier is in a broken state看看势垒是否处于破碎状态
        if self._state < 0:
            raise BrokenBarrierError
        assert self._state == 0

    # Optionally run the 'action' and release the threads waiting
    # in the barrier.
    #可以选择运行“action”，并释放等待在barrier中的线程。

    def _release(self):
        try:
            if self._action:
                self._action()
            # enter draining state  进入排水状态
            self._state = 1
            self._cond.notify_all()
        except:
            #an exception during the _action handler.  Break and reraise
            #_action处理程序期间的异常。打破和reraise
            self._break()
            raise

    # Wait in the barrier until we are released.  Raise an exception
    # if the barrier is reset or broken.
    #在障碍物里等着，直到我们被释放。如果障碍被重置或破坏，则引发异常。
    def _wait(self, timeout):
        if not self._cond.wait_for(lambda : self._state != 0, timeout):
            #timed out.  Break the barrier
            self._break()
            raise BrokenBarrierError
        if self._state < 0:
            raise BrokenBarrierError
        assert self._state == 1

    # If we are the last thread to exit the barrier, signal any threads
    #     # waiting for the barrier to drain.
    #如果我们是最后一个退出屏障的线程，那么向等待屏障流出的线程发出信号。
    def _exit(self):
        if self._count == 0:
            if self._state in (-1, 1):
                #resetting or draining
                self._state = 0
                self._cond.notify_all()

    def reset(self):
        """Reset the barrier to the initial state.
        Any threads currently waiting will get the BrokenBarrier exception
        raised.
        """
        #将势垒重置为初始状态。当前等待的任何线程都将引发故障障碍异常。
        with self._cond:
            if self._count > 0:
                if self._state == 0:
                    #reset the barrier, waking up threads 重置障碍，唤醒线程
                    self._state = -1
                elif self._state == -2:
                    #was broken, set it to reset state 被破坏，设置为重置状态
                    #which clears when the last thread exits 最后一个线程退出时哪个线程清除
                    self._state = -1
            else:
                self._state = 0
            self._cond.notify_all()

    def abort(self):
        """Place the barrier into a 'broken' state.
        Useful in case of error.  Any currently waiting threads and threads
        attempting to 'wait()' will have BrokenBarrierError raised.
        """
        #将障碍设置为“破碎”状态。在发生错误时很有用。任何当前正在等待的线程和
        # 试图“wait()”的线程都会出现故障障碍。
        with self._cond:
            self._break()

    def _break(self):
        # An internal error was detected.  The barrier is set to
        # a broken state all parties awakened.
        #检测到内部错误。障碍被设置为一个破碎的国家，所有各方都觉醒了。
        self._state = -2
        self._cond.notify_all()

    @property
    def parties(self):
        """Return the number of threads required to trip the barrier."""
        #返回跳闸所需的线程数。
        return self._parties

    @property
    def n_waiting(self):
        """Return the number of threads currently waiting at the barrier."""
        #返回阻塞处当前等待的线程数。
        # We don't need synchronization here since this is an ephemeral result
        # anyway.  It returns the correct value in the steady state.
        #我们不需要同步，因为这是一个短暂的结果。它在稳定状态下返回正确的值。
        if self._state == 0:
            return self._count
        return 0

    @property
    def broken(self):
        """Return True if the barrier is in a broken state."""
        #如果屏障处于破坏状态，返回True。
        return self._state == -2

# exception raised by the Barrier class
#由Barrier类引发的异常
class BrokenBarrierError(RuntimeError):
    pass


# Helper to generate new thread names
#帮助程序生成新的线程名称
_counter = _count().__next__
_counter() # Consume 0 so first non-main thread has id 1.
           #消耗0，所以第一个非主线程id为1。
def _newname(template="Thread-%d"):
    return template % _counter()

# Active thread administration  #活动线程管理
_active_limbo_lock = _allocate_lock()
_active = {}    # maps thread id to Thread object 将线程id映射到线程对象
_limbo = {}
_dangling = WeakSet()

# Main class for threads
'''线程的主类'''

class Thread:
    """A class that represents a thread of control.
    This class can be safely subclassed in a limited fashion. There are two ways
    to specify the activity: by passing a callable object to the constructor, or
    by overriding the run() method in a subclass.
    """
    #表示控制线程的类。这个类可以以有限的方式安全地子类化。有两种方法可以指定活动:
    # 通过将可调用对象传递给构造函数，或者在子类中重写run()方法。

    _initialized = False
    # Need to store a reference to sys.exc_info for printing
    # out exceptions when a thread tries to use a global var. during interp.
    # shutdown and thus raises an exception about trying to perform some
    # operation on/with a NoneType
    #需要存储对sys的引用。exc_info用于在interp期间线程试图使用全局变量时打印异常。
    # 关闭，因此引发了一个异常，即试图对/使用非etype执行某些操作
    _exc_info = _sys.exc_info
    # Keep sys.exc_clear too to clear the exception just before
    # allowing .join() to return.
    #Keep sys.ex_clear也可以在allowing.join()返回之前清除异常。
    #XXX __exc_clear = _sys.exc_clear

    def __init__(self, group=None, target=None, name=None,
                 args=(), kwargs=None, *, daemon=None):
        """This constructor should always be called with keyword arguments. Arguments are:
        *group* should be None; reserved for future extension when a ThreadGroup
        class is implemented.
        *target* is the callable object to be invoked by the run()
        method. Defaults to None, meaning nothing is called.
        *name* is the thread name. By default, a unique name is constructed of
        the form "Thread-N" where N is a small decimal number.
        *args* is the argument tuple for the target invocation. Defaults to ().
        *kwargs* is a dictionary of keyword arguments for the target
        invocation. Defaults to {}.
        If a subclass overrides the constructor, it must make sure to invoke
        the base class constructor (Thread.__init__()) before doing anything
        else to the thread.
        """
        #这个构造函数应该总是使用关键字参数调用。论点是:*group*不应该是;在实现
        # ThreadGroup类时为将来的扩展保留。*target*是run()方法调用的可调用对象。
        # 默认为None，表示不调用任何东西。*name*是线程名。默认情况下，唯一的名称
        # 是由“Thread-N”的形式构造的，其中N是一个小数。*args*是目标调用的参数元组。
        # 默认为()。*kwargs*是目标调用的关键字参数字典。默认为{}。如果子类重写构造
        # 函数，它必须确保在对线程执行其他操作之前调用基类构造函数(thread. __init__())。

        assert group is None, "group argument must be None for now"
        if kwargs is None:
            kwargs = {}
        self._target = target
        self._name = str(name or _newname())
        self._args = args
        self._kwargs = kwargs
        if daemon is not None:
            self._daemonic = daemon
        else:
            self._daemonic = current_thread().daemon
        self._ident = None
        self._tstate_lock = None
        self._started = Event()
        self._is_stopped = False
        self._initialized = True
        # sys.stderr is not stored in the class like
        # sys.exc_info since it can be changed between instances
        self._stderr = _sys.stderr
        # For debugging and _after_fork()
        _dangling.add(self)

    def _reset_internal_locks(self, is_alive):
        # private!  Called by _after_fork() to reset our internal locks as
        # they may be in an invalid state leading to a deadlock or crash.
        #私人!由_after_fork()调用，以重置内部锁，因为它们可能处于无效状态，导致死锁或崩溃。
        self._started._reset_internal_locks()
        if is_alive:
            self._set_tstate_lock()
        else:
            # The thread isn't alive after fork: it doesn't have a tstate anymore.
            #在fork之后，线程不再是活的:它不再有tstate。
            self._is_stopped = True
            self._tstate_lock = None

    def __repr__(self):
        assert self._initialized, "Thread.__init__() was not called"
        status = "initial"
        if self._started.is_set():
            status = "started"
        self.is_alive() # easy way to get ._is_stopped set when appropriate
                        #在适当的情况下，获得._is_stopped设置的简单方法
        if self._is_stopped:
            status = "stopped"
        if self._daemonic:
            status += " daemon"
        if self._ident is not None:
            status += " %s" % self._ident
        return "<%s(%s, %s)>" % (self.__class__.__name__, self._name, status)

    def start(self):
        """Start the thread's activity.
        It must be called at most once per thread object. It arranges for the
        object's run() method to be invoked in a separate thread of control.
        This method will raise a RuntimeError if called more than once on the
        same thread object.
        """
        #启动线程的活动。每个线程对象最多只能调用一次。它安排在一个单独的控制线程中
        # 调用对象的run()方法。如果在同一个线程对象上调用多次，此方法将引发运行时错误。
        if not self._initialized:
            raise RuntimeError("thread.__init__() not called")

        if self._started.is_set():
            raise RuntimeError("threads can only be started once")
        with _active_limbo_lock:
            _limbo[self] = self
        try:
            _start_new_thread(self._bootstrap, ())
        except Exception:
            with _active_limbo_lock:
                del _limbo[self]
            raise
        self._started.wait()

    def run(self):
        """Method representing the thread's activity.
        You may override this method in a subclass. The standard run() method
        invokes the callable object passed to the object's constructor as the
        target argument, if any, with sequential and keyword arguments taken
        from the args and kwargs arguments, respectively.
        """
        #表示线程活动的方法。您可以在子类中重写此方法。标准run()方法调用传递给对象
        # 构造函数的可调用对象作为目标参数(如果有的话)，分别使用args和kwargs参数
        # 中的顺序参数和关键字参数。
        try:
            if self._target:
                self._target(*self._args, **self._kwargs)
        finally:
            # Avoid a refcycle if the thread is running a function with
            # an argument that has a member that points to the thread.
            #如果线程正在运行一个具有指向线程的成员的参数的函数，请避免使用refcycle。
            del self._target, self._args, self._kwargs

    def _bootstrap(self):
        # Wrapper around the real bootstrap code that ignores
        # exceptions during interpreter cleanup.  Those typically
        # happen when a daemon thread wakes up at an unfortunate
        # moment, finds the world around it destroyed, and raises some
        # random exception *** while trying to report the exception in
        # _bootstrap_inner() below ***.  Those random exceptions
        # don't help anybody, and they confuse users, so we suppress
        # them.  We suppress them only when it appears that the world
        # indeed has already been destroyed, so that exceptions in
        # _bootstrap_inner() during normal business hours are properly
        # reported.  Also, we only suppress them for daemonic threads;
        # if a non-daemonic encounters this, something else is wrong.
        '''包装真正的引导代码，在解释器清理期间忽略异常。这通常发生在守护进程线程
        在一个不幸的时刻醒来，发现它周围的世界被破坏，并在试图报告***下面的异常
        in_bootstrap_inner()时引发一些随机异常时。这些随机的异常对任何人都没有
        帮助，而且它们混淆了用户，所以我们抑制了它们。只有当世界似乎确实已经被破坏
        时，我们才会抑制它们，以便在正常工作时间内正确报告_bootstrap_inner()中
        的异常。而且，我们只对daemonic线程禁止它们;如果一个非daemonic遇到了这个
        问题，就会出现其他问题'''
        try:
            self._bootstrap_inner()
        except:
            if self._daemonic and _sys is None:
                return
            raise

    def _set_ident(self):
        self._ident = get_ident()

    def _set_tstate_lock(self):
        """
        Set a lock object which will be released by the interpreter when
        the underlying thread state (see pystate.h) gets deleted.
        """
        #设置一个锁对象，当底层线程状态(请参阅pystate.h)被删除时，解释器将释放这个锁对象。
        self._tstate_lock = _set_sentinel()
        self._tstate_lock.acquire()

    def _bootstrap_inner(self):
        try:
            self._set_ident()
            self._set_tstate_lock()
            self._started.set()
            with _active_limbo_lock:
                _active[self._ident] = self
                del _limbo[self]

            if _trace_hook:
                _sys.settrace(_trace_hook)
            if _profile_hook:
                _sys.setprofile(_profile_hook)

            try:
                self.run()
            except SystemExit:
                pass
            except:
                # If sys.stderr is no more (most likely from interpreter
                # shutdown) use self._stderr.  Otherwise still use sys (as in
                # _sys) in case sys.stderr was redefined since the creation of
                # self.
                #如果系统。stderr不再使用self._stderr(很可能是由于解释器关闭)。否则，
                # 在case sys中仍然使用sys(如in_sys)。stderr自自我创造以来被重新定义。
                if _sys and _sys.stderr is not None:
                    print("Exception in thread %s:
%s" %
                          (self.name, _format_exc()), file=_sys.stderr)
                elif self._stderr is not None:
                    # Do the best job possible w/o a huge amt. of code to
                    # approximate a traceback (code ideas from Lib/traceback.py)
                    #尽最大的努力做最好的工作。近似回溯的代码(来自Lib/traceback.py的代码思想)
                    exc_type, exc_value, exc_tb = self._exc_info()
                    try:
                        print((
                            "Exception in thread " + self.name +
                            " (most likely raised during interpreter shutdown):"), file=self._stderr)
                        print((
                            "Traceback (most recent call last):"), file=self._stderr)
                        while exc_tb:
                            print((
                                '  File "%s", line %s, in %s' %
                                (exc_tb.tb_frame.f_code.co_filename,
                                    exc_tb.tb_lineno,
                                    exc_tb.tb_frame.f_code.co_name)), file=self._stderr)
                            exc_tb = exc_tb.tb_next
                        print(("%s: %s" % (exc_type, exc_value)), file=self._stderr)
                        self._stderr.flush()
                    # Make sure that exc_tb gets deleted since it is a memory
                    # hog; deleting everything else is just for thoroughness
                    #确保exc_tb被删除，因为它占用内存;删除所有其他内容只是为了彻底
                    finally:
                        del exc_type, exc_value, exc_tb
            finally:
                # Prevent a race in
                # test_threading.test_no_refcycle_through_target when
                # the exception keeps the target alive past when we
                # assert that it's dead.
                #防止test_threading中的竞争。test_no_refcycle_through_target，
                # 当异常断言目标已死时，该异常将使目标保持存活。
                #XXX self._exc_clear()
                pass
        finally:
            with _active_limbo_lock:
                try:
                    # We don't call self._delete() because it also
                    # grabs _active_limbo_lock.
                    #我们不调用self._delete()，因为它也抓取_active_limbo_lock。
                    del _active[get_ident()]
                except:
                    pass

    def _stop(self):
        # After calling ._stop(), .is_alive() returns False and .join() returns
        # immediately.  ._tstate_lock must be released before calling ._stop().
        #调用._stop()后，.is_alive()返回False， .join()立即返回。

        # Normal case:  C code at the end of the thread's life
        # (release_sentinel in _threadmodule.c) releases ._tstate_lock, and
        # that's detected by our ._wait_for_tstate_lock(), called by .join()
        # and .is_alive().  Any number of threads _may_ call ._stop()
        # simultaneously (for example, if multiple threads are blocked in
        # .join() calls), and they're not serialized.  That's harmless -
        # they'll just make redundant rebindings of ._is_stopped and
        # ._tstate_lock.  Obscure:  we rebind ._tstate_lock last so that the
        # "assert self._is_stopped" in ._wait_for_tstate_lock() always works
        # (the assert is executed only if ._tstate_lock is None).
        #正常情况:线程生命周期结束时的C代码(_threadmodule.c中的release_sentinel)
        # 释放了._tstate_lock，我们的._wait_for_tstate_lock()检测到这一点，
        # 它被.join()和.is_alive()调用。同时调用任意数量的线程_may_ ._stop()
        # (例如，如果多个线程在.join()调用中被阻塞，并且它们没有被序列化)。这是无害的，
        # 他们只会对._is_stopped和._tstate_lock进行冗余的重绑定。晦涩的:
        # 我们将._tstate_lock绑定到最后，以便“断言self”。_is_stopped()中
        # 的._wait_for_tstate_lock()总是有效的(只有当._tstate_lock为空时才执行断言)。

        # Special case:  _main_thread releases ._tstate_lock via this
        # module's _shutdown() function.
        #特殊情况:_main_thread通过这个模块的_shutdown()函数释放._tstate_lock。
        lock = self._tstate_lock
        if lock is not None:
            assert not lock.locked()
        self._is_stopped = True
        self._tstate_lock = None

    def _delete(self):
        "Remove current thread from the dict of currently running threads."
        with _active_limbo_lock:
            del _active[get_ident()]
            # There must not be any python code between the previous line
            # and after the lock is released.  Otherwise a tracing function
            # could try to acquire the lock again in the same thread, (in
            # current_thread()), and would block.
            #前一行和锁释放后之间不应该有任何python代码。否则，跟踪函数可以尝试在相
            # 同的线程(在current_thread()中)中再次获取锁，并将阻塞。

    def join(self, timeout=None):
        """Wait until the thread terminates.
        This blocks the calling thread until the thread whose join() method is
        called terminates -- either normally or through an unhandled exception
        or until the optional timeout occurs.
        When the timeout argument is present and not None, it should be a
        floating point number specifying a timeout for the operation in seconds
        (or fractions thereof). As join() always returns None, you must call
        isAlive() after join() to decide whether a timeout happened -- if the
        thread is still alive, the join() call timed out.
        When the timeout argument is not present or None, the operation will
        block until the thread terminates.
        A thread can be join()ed many times.
        join() raises a RuntimeError if an attempt is made to join the current
        thread as that would cause a deadlock. It is also an error to join() a
        thread before it has been started and attempts to do so raises the same
        exception.
        """
        #等待直到线程终止。这将阻塞调用线程，直到调用join()方法的线程终止——通常或通过
        # 未处理的异常终止，或直到出现可选超时为止。当出现timeout参数而不是None时，
        # 它应该是一个浮点数，以秒(或几分之一)为单位指定操作的超时。因为join()总是
        # 返回None，所以必须在join()之后调用isAlive()，以决定是否发生超时——如果线程
        # 仍然活着，则join()调用超时。当timeout参数不存在或不存在时，操作将阻塞，
        # 直到线程终止。一个线程可以多次连接()ed。如果尝试连接当前线程，join()将引发
        # 一个运行时错误，因为这会导致死锁。在线程启动之前连接()线程也是一个错误，
        # 试图这样做会引发相同的异常。
        if not self._initialized:
            raise RuntimeError("Thread.__init__() not called")
        if not self._started.is_set():
            raise RuntimeError("cannot join thread before it is started")
        if self is current_thread():
            raise RuntimeError("cannot join current thread")

        if timeout is None:
            self._wait_for_tstate_lock()
        else:
            # the behavior of a negative timeout isn't documented, but
            # historically .join(timeout=x) for x<0 has acted as if timeout=0
            #没有记录消极超时的行为，但是在历史上，x<0时的.join(timeout=x)就像timeout=0一样
            self._wait_for_tstate_lock(timeout=max(timeout, 0))

    def _wait_for_tstate_lock(self, block=True, timeout=-1):
        # Issue #18808: wait for the thread state to be gone.
        # At the end of the thread's life, after all knowledge of the thread
        # is removed from C data structures, C code releases our _tstate_lock.
        # This method passes its arguments to _tstate_lock.acquire().
        # If the lock is acquired, the C code is done, and self._stop() is
        # called.  That sets ._is_stopped to True, and ._tstate_lock to None.
        #问题#18808:等待线程状态消失。在线程生命周期结束时，在从C数据结构中删除所有
        # 线程知识之后，C代码释放我们的_tstate_lock。该方法将其参数
        # 传递给_tstate_lock.acquire()。如果获得了锁，则完成C代码，
        # 并调用self._stop()。这将._is_stopped设置为True，._tstate_lock设置为None。
        lock = self._tstate_lock
        if lock is None:  # already determined that the C code is done 已经确定C代码已经完成
            assert self._is_stopped
        elif lock.acquire(block, timeout):
            lock.release()
            self._stop()

    @property
    def name(self):
        """A string used for identification purposes only.
        It has no semantics. Multiple threads may be given the same name. The
        initial name is set by the constructor.
        """
        #仅用于识别目的的字符串。它没有语义。多个线程可能被赋予相同的名称。初始名称由构造函数设置。
        assert self._initialized, "Thread.__init__() not called"
        return self._name

    @name.setter
    def name(self, name):
        assert self._initialized, "Thread.__init__() not called"
        self._name = str(name)

    @property
    def ident(self):
        """Thread identifier of this thread or None if it has not been started.
        This is a nonzero integer. See the get_ident() function. Thread
        identifiers may be recycled when a thread exits and another thread is
        created. The identifier is available even after the thread has exited.
        """
        #此线程的线程标识符，如果没有启动，则为空。这是非零整数。请参阅get_ident()函数。
        # 当线程退出并创建另一个线程时，可以回收线程标识符。即使线程已经退出，标识符也是可用的。
        assert self._initialized, "Thread.__init__() not called"
        return self._ident

    def is_alive(self):
        """Return whether the thread is alive.
        This method returns True just before the run() method starts until just
        after the run() method terminates. The module function enumerate()
        returns a list of all alive threads.
        """
        #返回线程是否存在。这个方法在run()方法开始之前返回True，直到run()方法终止之后。
        # 模块函数enumerate()返回一个包含所有活线程的列表。
        assert self._initialized, "Thread.__init__() not called"
        if self._is_stopped or not self._started.is_set():
            return False
        self._wait_for_tstate_lock(False)
        return not self._is_stopped

    isAlive = is_alive

    @property
    def daemon(self):
        """A boolean value indicating whether this thread is a daemon thread.
        This must be set before start() is called, otherwise RuntimeError is
        raised. Its initial value is inherited from the creating thread; the
        main thread is not a daemon thread and therefore all threads created in
        the main thread default to daemon = False.
        The entire Python program exits when no alive non-daemon threads are
        left.
        """
        #一个布尔值，指示此线程是否为守护线程。这必须在调用start()之前设置，否则会引发
        # 运行时错误。它的初始值继承自创建线程;主线程不是守护进程线程，因此在主线程中
        # 创建的所有线程默认为守护进程= False。当没有存活的非守护进程线程时，
        # 整个Python程序退出。
        assert self._initialized, "Thread.__init__() not called"
        return self._daemonic

    @daemon.setter
    def daemon(self, daemonic):
        if not self._initialized:
            raise RuntimeError("Thread.__init__() not called")
        if self._started.is_set():
            raise RuntimeError("cannot set daemon status of active thread")
        self._daemonic = daemonic

    def isDaemon(self):   #Daemon:守护进程
        return self.daemon

    def setDaemon(self, daemonic):
        self.daemon = daemonic

    def getName(self):
        return self.name

    def setName(self, name):
        self.name = name

# The timer class was contributed by Itamar Shtull-Trauring
#计时器类由Itamar Shtull-Trauring贡献

class Timer(Thread):
    """Call a function after a specified number of seconds:
            t = Timer(30.0, f, args=None, kwargs=None)
            t.start()
            t.cancel()     # stop the timer's action if it's still waiting
    """
    #在指定的秒数后调用一个函数:t = Timer(30.0, f, args=None, kwargs=None)
    #t.start() t.cancel()如果计时器仍在等待，则停止计时器的操作

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

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()


# Special thread class to represent the main thread
'''表示主线程的特殊线程类'''

class _MainThread(Thread):

    def __init__(self):
        Thread.__init__(self, name="MainThread", daemon=False)
        self._set_tstate_lock()
        self._started.set()
        self._set_ident()
        with _active_limbo_lock:
            _active[self._ident] = self


# Dummy thread class to represent threads not started here.
# These aren't garbage collected when they die, nor can they be waited for.
# If they invoke anything in threading.py that calls current_thread(), they
# leave an entry in the _active dict forever after.
# Their purpose is to return *something* from current_thread().
# They are marked as daemon threads so we won't wait for them
# when we exit (conform previous semantics).
#伪线程类来表示这里没有启动的线程。它们死后不会被垃圾收集，也不会被等待。如果它们在
# 线程中调用任何东西。调用current_thread()的py在_active dict中永远留下一个条目。
# 它们的目的是从current_thread()返回*something*。它们被标记为守护线程，因此在退出
# 时我们不会等待它们(符合前面的语义)。

class _DummyThread(Thread):

    def __init__(self):
        Thread.__init__(self, name=_newname("Dummy-%d"), daemon=True)

        self._started.set()
        self._set_ident()
        with _active_limbo_lock:
            _active[self._ident] = self

    def _stop(self):
        pass

    def is_alive(self):
        assert not self._is_stopped and self._started.is_set()
        return True

    def join(self, timeout=None):
        assert False, "cannot join a dummy thread"


# Global API functions
#全球API函数

def current_thread():
    """Return the current Thread object, corresponding to the caller's thread of control.
    If the caller's thread of control was not created through the threading
    module, a dummy thread object with limited functionality is returned.
    """
    #返回当前线程对象，对应于调用方的控制线程。如果没有通过线程模块创建调用者的控制线
    # 程，则返回具有有限功能的虚拟线程对象。
    try:
        return _active[get_ident()]
    except KeyError:
        return _DummyThread()

currentThread = current_thread

def active_count():
    """Return the number of Thread objects currently alive.
    The returned count is equal to the length of the list returned by
    enumerate().
    """
    #返回当前存活的线程对象的数量。返回的计数等于enumerate()返回的列表的长度。
    with _active_limbo_lock:
        return len(_active) + len(_limbo)

activeCount = active_count

def _enumerate():
    # Same as enumerate(), but without the lock. Internal use only.
    #与enumerate()相同，只是没有锁。内部使用。
    return list(_active.values()) + list(_limbo.values())

def enumerate():
    """Return a list of all Thread objects currently alive.
    The list includes daemonic threads, dummy thread objects created by
    current_thread(), and the main thread. It excludes terminated threads and
    threads that have not yet been started.
    """
    #返回当前所有线程对象的列表。该列表包括daemonic线程、current_thread()创建的虚拟
    # 线程对象和主线程。它排除终止的线程和尚未启动的线程。
    with _active_limbo_lock:
        return list(_active.values()) + list(_limbo.values())

from _thread import stack_size

# Create the main thread object,
# and make it available for the interpreter
# (Py_Main) as threading._shutdown.
#创建主线程对象，并将其作为thread ._shutdown提供给解释器(Py_Main)。

_main_thread = _MainThread()

def _shutdown():
    # Obscure:  other threads may be waiting to join _main_thread.  That's
    # dubious, but some code does it.  We can't wait for C code to release
    # the main thread's tstate_lock - that won't happen until the interpreter
    # is nearly dead.  So we release it here.  Note that just calling _stop()
    # isn't enough:  other threads may already be waiting on _tstate_lock.
    #晦涩:其他线程可能正在等待加入_main_thread。这很可疑，但有些代码可以做到。
    # 我们不能等待C代码释放主线程的tstate_lock——这要等到解释器快死的时候才会发生。
    # 我们在这里释放它。注意，仅仅调用_stop()是不够的:其他线程可能已经在
    # 等待_tstate_lock了。
    if _main_thread._is_stopped:
        # _shutdown() was already called
        return
    tlock = _main_thread._tstate_lock
    # The main thread isn't finished yet, so its thread state lock can't have
    # been released.
    #主线程尚未完成，因此它的线程状态锁无法释放。
    assert tlock is not None
    assert tlock.locked()
    tlock.release()
    _main_thread._stop()
    t = _pickSomeNonDaemonThread()
    while t:
        t.join()
        t = _pickSomeNonDaemonThread()

def _pickSomeNonDaemonThread():
    for t in enumerate():
        if not t.daemon and t.is_alive():
            return t
    return None

def main_thread():
    """Return the main thread object.
    In normal conditions, the main thread is the thread from which the
    Python interpreter was started.
    """
    #返回主线程对象。在正常情况下，主线程是Python解释器启动的线程。
    return _main_thread

# get thread-local implementation, either from the thread
# module, or from the python fallback
#从线程模块或python回退中获取线程本地实现

try:
    from _thread import _local as local
except ImportError:
    from _threading_local import local


def _after_fork():
    """
    Cleanup threading module state that should not exist after a fork.
    """
    # Reset _active_limbo_lock, in case we forked while the lock was held
    # by another (non-forked) thread.  http://bugs.python.org/issue874900
    #Reset _active_limbo_lock,以防我们分叉而锁被另一个(非分叉的)线程持有。
    global _active_limbo_lock, _main_thread
    _active_limbo_lock = _allocate_lock()

    # fork() only copied the current thread; clear references to others.
    #fork()只复制当前线程;明确提及他人。
    new_active = {}
    current = current_thread()
    _main_thread = current
    with _active_limbo_lock:
        # Dangling thread instances must still have their locks reset,
        # because someone may join() them.
        #悬空线程实例必须重新设置它们的锁，因为有人可能会加入()它们。
        threads = set(_enumerate())
        threads.update(_dangling)
        for thread in threads:
            # Any lock/condition variable may be currently locked or in an
            # invalid state, so we reinitialize them.
            #任何锁/条件变量可能当前被锁定或处于无效状态，因此我们重新初始化它们。
            if thread is current:
                # There is only one active thread. We reset the ident to
                # its new value since it can have changed.
                #只有一个活动线程。我们将ident重置为它的新值，因为它可能已经更改。
                thread._reset_internal_locks(True)
                ident = get_ident()
                thread._ident = ident
                new_active[ident] = thread
            else:
                # All the others are already stopped.
                thread._reset_internal_locks(False)
                thread._stop()

        _limbo.clear()
        _active.clear()
        _active.update(new_active)
        assert len(_active) == 1


if hasattr(_os, "register_at_fork"):
    _os.register_at_fork(after_in_child=_after_fork)

　　队列：

　　Python的Queue模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列PriorityQueue。这些队列都实现了锁原语，能够在多线程中直接使用。可以使用队列来实现线程间的同步。

'''队列'''
import queue
q =queue.Queue()    #设置队列
q.put("q1")              #队列中放入数据
q.put("q2")
q.put("q3")

# print(q.qsize())  #获取队列大小

'''队列中获取数据，取出的数据超出存入数据时会等待，不会报错'''
print(q.get())
print(q.get())
print(q.get())
# print(q.get())

'''获取队列，但不会等待，超出后直接报错'''
print(q.get_nowait())
print(q.get_nowait())
print(q.get_nowait())
# print(q.get_nowait())

'''设置优先级排序的依据'''
q = queue.PriorityQueue(maxsize=0)
q.put((3,"q1"))        #当maxsizie<=0时，队列无限大，>0时，给定数据即为队列大小
q.put((1,"q2"))
q.put((-4,"q3"))
print(q.get())         #获取时会从小到大按顺序获取
print(q.get())
print(q.get())

上述代码只是队列的应用，下面将队列应用与线程之中：

import queue
import time
import threading

q = queue.Queue(maxsize=10)
def gave(name):
    count = 1
    while True:
        q.put("--骨头--%s" % count)
        print("%s 生产骨头 %s" % (name,count))
        time.sleep(1)
        count+=1

def consumer(name):
    while q.qsize()>0:
    # while True:
        print("%s 吃掉 %s" % (name,q.get()))
        # time.sleep(10)

g = threading.Thread(target=gave,args=("王二小",))
c = threading.Thread(target=consumer,args=("旺财",))
g.start()
c.start()

#print('33[41;1m--red light on---33[0m')    #红灯
#print('33[43;1m--yellow light on---33[0m') #黄灯
#print('33[42;1m--green light on---33[0m') #绿灯
'''主要用在数据同步上'''
'''红绿灯事件'''

# import threading
# import time
# # import queue
# event = threading.Event()
# # q = queue.Queue()
#
# def light():
#     count = 1
#     while True:
#         if count<=5:
#             event.set()
#             print('33[42;1m--green light on---33[0m')
#         elif 5<count<=10:
#             event.clear()
#             print('33[43;1m--yellow light on---33[0m')
#         else:
#             print('33[41;1m--red light on---33[0m')
#             if count>=15:
#                 count = 0
#         time.sleep(1)
#         count+=1
#
# def car(name):
#     while True:
#         if event.is_set():
#             time.sleep(1)
#             print("%s is running..." % name)
#         else:
#             print("car is waiting...")
#             event.wait()  #等待事件event对象发生变化
#
#
# Light = threading.Thread(target=light,)
# Light.start()
# Car = threading.Thread(target=car,args=("BENZ",))
# Car.start()


import threading
import time
import queue

event=threading.Event()
q=queue.PriorityQueue(maxsize=20)
#在循环之前先放入十辆车：
for i in range(10):
    q.put("旧车辆，%s" % "QQ")

def light():
    count=0
    while True:
        if count<10:
            event.set()
            print("33[42;1m--green light on---33[0m",10-count)
        elif 10<=count<15:
            event.clear()
            print("33[43;1m--yellow light on---33[0m",15-count)
        else:
            event.clear()
            if count>=25:
                count=0
                continue
            print("33[41;1m--red light on---33[0m",25-count)
        time.sleep(1)
        count+=1

def car(name):
    while True:
        if event.is_set() and q.qsize()>=1:
            print("%s is running..." % name)
            time.sleep(1)
            print("道路还有【%s】辆车" % q.qsize())
        else:
            print("car is waiting...")
            print("现在道路中有车%s辆" % q.qsize())
            event.wait()   #等待事件event对象发生变化

#路口停车
def Put():
    n=0
    while q.qsize()<20:
        time.sleep(2)
        q.put("新车辆%s车辆" % n)
        n+=1
        print("车辆已驶入")
    else:
        event.wait()
        print("停止驶入")
        print("停止驶入后道路中有车%s" % q.qsize())

#车辆行驶
def Get():
    while True:
        if event.is_set():
            time.sleep(2)
            print("%s车辆--------通过" % q.get())
        else:
            print("禁止通行！！")
            event.wait()



C=threading.Thread(target=car,args=("...T...",))
L=threading.Thread(target=light)
P=threading.Thread(target=Put)
G=threading.Thread(target=Get)
L.start()
C.start()
P.start()
G.start()

import threading

money = 0
lock = threading.Lock()

#存钱
def get_money(Sum):
    global money
    money+=Sum     #x=money+sum;money=x

#取钱
def put_money(Sum):
    global money
    money-=Sum

def run(Sum):
    lock.acquire()
    for i in range(10000):
        put_money(Sum)
        get_money(Sum)
    lock.release()

#单线程中不会存在问题
#然而在多线程中，操作系统交叉处理赋值语句，导致
# 全局变量被一个线程修改，而另一个线程却不知情。
m1 = threading.Thread(target=run,args=(100,))
m2 = threading.Thread(target=run,args=(1000,))
m1.start()
m2.start()
m1.join()
m2.join()
print(money)

'''A multi-producer, multi-consumer queue.'''
#多生产者、多消费者队列。
import threading
from collections import deque
from heapq import heappush, heappop
from time import monotonic as time
try:
    from _queue import SimpleQueue
except ImportError:
    SimpleQueue = None

__all__ = ['Empty', 'Full', 'Queue', 'PriorityQueue', 'LifoQueue', 'SimpleQueue']


try:
    from _queue import Empty
except AttributeError:
    class Empty(Exception):
        'Exception raised by Queue.get(block=0)/get_nowait().'
        pass

class Full(Exception):
    'Exception raised by Queue.put(block=0)/put_nowait().'
    pass


class Queue:
    '''Create a queue object with a given maximum size.
    If maxsize is <= 0, the queue size is infinite.
    '''
    #创建一个具有给定最大大小的队列对象。如果maxsize <= 0，则队列大小为无穷大。

    def __init__(self, maxsize=0):
        self.maxsize = maxsize
        self._init(maxsize)

        # mutex must be held whenever the queue is mutating.  All methods
        # that acquire mutex must release it before returning.  mutex
        # is shared between the three conditions, so acquiring and
        # releasing the conditions also acquires and releases mutex.
        #当队列发生变化时，必须持有互斥锁。所有获得互斥锁的方法都必须在返回之前释放它。
        # 互斥锁在这三个条件之间是共享的，因此获取和释放条件也获得和释放互斥锁。
        self.mutex = threading.Lock()

        # Notify not_empty whenever an item is added to the queue; a
        # thread waiting to get is notified then.
        #当一个项目被添加到队列中时，通知not_empty;然后会通知等待获取的线程。
        self.not_empty = threading.Condition(self.mutex)

        # Notify not_full whenever an item is removed from the queue;
        # a thread waiting to put is notified then.
        #当一个项目从队列中删除时，通知not_full;然后会通知等待放置的线程。
        self.not_full = threading.Condition(self.mutex)

        # Notify all_tasks_done whenever the number of unfinished tasks
        # drops to zero; thread waiting to join() is notified to resume
        #当未完成任务的数量降为零时，通知all_tasks_done;等待加入()的线程被通知恢复
        self.all_tasks_done = threading.Condition(self.mutex)
        self.unfinished_tasks = 0

    def task_done(self):
        '''Indicate that a formerly enqueued task is complete.
        Used by Queue consumer threads.  For each get() used to fetch a task,
        a subsequent call to task_done() tells the queue that the processing
        on the task is complete.
        If a join() is currently blocking, it will resume when all items
        have been processed (meaning that a task_done() call was received
        for every item that had been put() into the queue).
        Raises a ValueError if called more times than there were items
        placed in the queue.
        '''
        #指示以前加入队列的任务已经完成。由队列使用者线程使用。对于用于获取任务的每个
        # get()，对task_done()的后续调用将告诉队列任务的处理已经完成。如果一个join()
        # 当前处于阻塞状态，那么当所有项都被处理完时(这意味着对于每个已将()放入队列的
        # 项都接收了task_done()调用)，它将恢复。如果调用的次数超过了队列中放置的项的
        # 次数，就会引发ValueError。
        with self.all_tasks_done:
            unfinished = self.unfinished_tasks - 1
            if unfinished <= 0:
                if unfinished < 0:
                    raise ValueError('task_done() called too many times')
                self.all_tasks_done.notify_all()
            self.unfinished_tasks = unfinished

    def join(self):
        '''Blocks until all items in the Queue have been gotten and processed.
        The count of unfinished tasks goes up whenever an item is added to the
        queue. The count goes down whenever a consumer thread calls task_done()
        to indicate the item was retrieved and all work on it is complete.
        When the count of unfinished tasks drops to zero, join() unblocks.
        '''
        #阻塞，直到获取和处理队列中的所有项。当一个项目被添加到队列中时，未完成任务的计数
        # 就会上升。每当使用者线程调用task_done()时，计数就会下降，以指示检索了项目并完成
        # 了对其的所有工作。当未完成任务的计数降为0时，join()将解块。
        with self.all_tasks_done:
            while self.unfinished_tasks:
                self.all_tasks_done.wait()

    def qsize(self):
        '''Return the approximate size of the queue (not reliable!).'''
        #返回队列的大致大小(不可靠!)
        with self.mutex:
            return self._qsize()

    def empty(self):
        '''Return True if the queue is empty, False otherwise (not reliable!).
        This method is likely to be removed at some point.  Use qsize() == 0
        as a direct substitute, but be aware that either approach risks a race
        condition where a queue can grow before the result of empty() or
        qsize() can be used.
        To create code that needs to wait for all queued tasks to be
        completed, the preferred technique is to use the join() method.
        '''
        #如果队列为空，返回True，否则返回False(不可靠!)这种方法可能会在某个时候被删除。
        # 使用qsize() == 0作为直接的替代，但是要注意，在使用empty()或qsize()的结果之前，
        # 队列可能会增长，这可能会带来竞争条件的风险。要创建需要等待所有排队任务完成的代码，
        # 首选技术是使用join()方法。
        with self.mutex:
            return not self._qsize()

    def full(self):
        '''Return True if the queue is full, False otherwise (not reliable!).
        This method is likely to be removed at some point.  Use qsize() >= n
        as a direct substitute, but be aware that either approach risks a race
        condition where a queue can shrink before the result of full() or
        qsize() can be used.
        '''
        #如果队列满了，返回True，否则返回False(不可靠!)这种方法可能会在某个时候被删除。
        # 使用qsize() >= n作为直接替代，但是要注意，在使用full()或qsize()的结果之前，
        # 队列可能会收缩，这可能会导致竞争条件的风险。
        with self.mutex:
            return 0 < self.maxsize <= self._qsize()

    def put(self, item, block=True, timeout=None):
        '''Put an item into the queue.
        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until a free slot is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Full exception if no free slot was available within that time.
        Otherwise ('block' is false), put an item on the queue if a free slot
        is immediately available, else raise the Full exception ('timeout'
        is ignored in that case).
        '''
        #将项目放入队列中。如果可选的args 'block'为true，而'timeout'为None(默认值)，
        # 那么如果有必要，阻塞直到空闲的插槽可用为止。如果“timeout”是非负数，它最多会
        # 阻塞“timeout”秒，如果在这段时间内没有可用的空闲时间，它就会引发完全异常。
        # 否则(‘block’为false)，如果有空闲的插槽立即可用，就在队列中放置一个项目，
        # 否则引发完整的异常(在这种情况下忽略‘timeout’)。
        with self.not_full:
            if self.maxsize > 0:
                if not block:
                    if self._qsize() >= self.maxsize:
                        raise Full
                elif timeout is None:
                    while self._qsize() >= self.maxsize:
                        self.not_full.wait()
                elif timeout < 0:
                    raise ValueError("'timeout' must be a non-negative number")
                else:
                    endtime = time() + timeout
                    while self._qsize() >= self.maxsize:
                        remaining = endtime - time()
                        if remaining <= 0.0:
                            raise Full
                        self.not_full.wait(remaining)
            self._put(item)
            self.unfinished_tasks += 1
            self.not_empty.notify()

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.
        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until an item is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Empty exception if no item was available within that time.
        Otherwise ('block' is false), return an item if one is immediately
        available, else raise the Empty exception ('timeout' is ignored
        in that case).
        '''
        #从队列中删除并返回项。如果可选的args 'block'为true，而'timeout'为None
        # (默认值)，则在项可用之前，如果有必要，阻塞。如果“timeout”是非负数，它最多
        # 会阻塞“timeout”秒，如果在这段时间内没有可用项，就会引发空异常。
        # 否则(‘block’为false)，如果一个项立即可用，返回一个项，否则引发空异常
        # (在这种情况下忽略'timeout')。
        with self.not_empty:
            if not block:
                if not self._qsize():
                    raise Empty
            elif timeout is None:
                while not self._qsize():
                    self.not_empty.wait()
            elif timeout < 0:
                raise ValueError("'timeout' must be a non-negative number")
            else:
                endtime = time() + timeout
                while not self._qsize():
                    remaining = endtime - time()
                    if remaining <= 0.0:
                        raise Empty
                    self.not_empty.wait(remaining)
            item = self._get()
            self.not_full.notify()
            return item

    def put_nowait(self, item):
        '''Put an item into the queue without blocking.
        Only enqueue the item if a free slot is immediately available.
        Otherwise raise the Full exception.
        '''
        #将项目放入队列中而不阻塞。只有当一个空闲的插槽立即可用时，才将项目加入队列。否则引发完全异常。
        return self.put(item, block=False)

    def get_nowait(self):
        '''Remove and return an item from the queue without blocking.
        Only get an item if one is immediately available. Otherwise
        raise the Empty exception.
        '''
        #在不阻塞的情况下从队列中删除并返回项。只有当一个项目是立即可用的。否则引发空异常。
        return self.get(block=False)

    # Override these methods to implement other queue organizations
    # (e.g. stack or priority queue).
    # These will only be called with appropriate locks held
    #重写这些方法以实现其他队列组织(例如堆栈或优先队列)。只有在持有适当的锁时才会调用这些函数

    # Initialize the queue representation
    '''初始化队列表示'''
    def _init(self, maxsize):
        self.queue = deque()

    def _qsize(self):
        return len(self.queue)

    # Put a new item in the queue
    def _put(self, item):
        self.queue.append(item)

    # Get an item from the queue
    def _get(self):
        return self.queue.popleft()


class PriorityQueue(Queue):
    '''Variant of Queue that retrieves open entries in priority order (lowest first).
    Entries are typically tuples of the form:  (priority number, data).
    '''
    #按优先级顺序(最低优先级)检索打开项的队列的变体。条目通常是表单的元组(优先级号、数据)。

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        heappush(self.queue, item)

    def _get(self):
        return heappop(self.queue)


class LifoQueue(Queue):
    '''Variant of Queue that retrieves most recently added entries first.'''
    #队列的变体，它首先检索最近添加的条目。

    def _init(self, maxsize):
        self.queue = []

    def _qsize(self):
        return len(self.queue)

    def _put(self, item):
        self.queue.append(item)

    def _get(self):
        return self.queue.pop()


class _PySimpleQueue:
    '''Simple, unbounded FIFO queue.
    This pure Python implementation is not reentrant.
    '''
    #简单、无界的FIFO队列。这个纯Python实现是不可重入的。

    # Note: while this pure Python version provides fairness
    # (by using a threading.Semaphore which is itself fair, being based
    #  on threading.Condition), fairness is not part of the API contract.
    # This allows the C version to use a different implementation.
    #注意:虽然这个纯Python版本提供了公平性(通过使用线程)。信号量本身是公平的，
    # 基于thread . condition)，公平不是API契约的一部分。这允许C版本使用不同的实现。

    def __init__(self):
        self._queue = deque()
        self._count = threading.Semaphore(0)

    def put(self, item, block=True, timeout=None):
        '''Put the item on the queue.
        The optional 'block' and 'timeout' arguments are ignored, as this method
        never blocks.  They are provided for compatibility with the Queue class.
        '''
        #将项目放到队列中。可选的“block”和“timeout”参数被忽略，因为这个方法从不阻塞。
        # 它们是为了与队列类兼容而提供的。
        self._queue.append(item)
        self._count.release()

    def get(self, block=True, timeout=None):
        '''Remove and return an item from the queue.
        If optional args 'block' is true and 'timeout' is None (the default),
        block if necessary until an item is available. If 'timeout' is
        a non-negative number, it blocks at most 'timeout' seconds and raises
        the Empty exception if no item was available within that time.
        Otherwise ('block' is false), return an item if one is immediately
        available, else raise the Empty exception ('timeout' is ignored
        in that case).
        '''
        #从队列中删除并返回项。如果可选的args 'block'为true，而'timeout'为None
        # (默认值)，则在项可用之前，如果有必要，阻塞。如果“timeout”是非负数，它最多
        # 会阻塞“timeout”秒，如果在这段时间内没有可用项，就会引发空异常。否则
        # (‘block’为false)，如果一个项立即可用，返回一个项，否则引发空异常
        # (在这种情况下忽略'timeout')。
        if timeout is not None and timeout < 0:
            raise ValueError("'timeout' must be a non-negative number")
        if not self._count.acquire(block, timeout):
            raise Empty
        return self._queue.popleft()

    def put_nowait(self, item):
        '''Put an item into the queue without blocking.
        This is exactly equivalent to `put(item)` and is only provided
        for compatibility with the Queue class.
        '''
        #将项目放入队列中而不阻塞。这完全等同于‘put(item)’，并且只提供与队列类的兼容性。
        return self.put(item, block=False)

    def get_nowait(self):
        '''Remove and return an item from the queue without blocking.
        Only get an item if one is immediately available. Otherwise
        raise the Empty exception.
        '''
        #在不阻塞的情况下从队列中删除并返回项。只有当一个项目是立即可用的。否则引发空异常。
        return self.get(block=False)

    def empty(self):
        '''Return True if the queue is empty, False otherwise (not reliable!).'''
        #如果队列为空，返回True，否则返回False(不可靠!)
        return len(self._queue) == 0

    def qsize(self):
        '''Return the approximate size of the queue (not reliable!).'''
        #返回队列的大致大小(不可靠!)
        return len(self._queue)


if SimpleQueue is None:
    SimpleQueue = _PySimpleQueue

Python进程

进程（multiprocessing）：

线程是进程最小的数据单元；每个进程都是相互独立的，它们之间不能共享数据。

启动单个进程：

'''启动一个进程'''
import multiprocessing
import time

def run(name):
    time.sleep(2)
    print("hello",name)

if __name__ == "__main__":
    p = multiprocessing.Process(target=run,args=("pp",))
    p.start()
    p.join()


'''运行结果'''
hello pp

启动多个进程：

import multiprocessing
import time

def run(name):
    time.sleep(2)
    print("hello",name)

for i in range(10):
    if __name__ == "__main__":
        p = multiprocessing.Process(target=run, args=("pp",))
        p.start()
        p.join()

在进程中创建进程：

import multiprocessing
import threading
import time

def thread_run():
    print(threading.get_ident())  #get_ident()获得线程地址

def run(name):
    time.sleep(2)
    print("hello",name)
    t = threading.Thread(target=thread_run)
    t.start()

if __name__ == "__main__":
    p = multiprocessing.Process(target=run,args=("pp",))
    p.start()
    p.join()


'''运行结果'''
hello pp
2404

数据共享：

'''通过中间介质（pickle）使两个进程实现数据共享,实质上并不是完全的数据共享，
只是将子进程的对象（队列Queue）进行克隆'''
# import threading
# import queue
# 
# def f():
#     q.put("jfkdsljfkdls")
# 
# if __name__=="__main__":
#     q=queue.Queue()
#     p=threading.Thread(target=f)
#     p.start()
#     print(q.get())
#     p.join()

'''进程'''
from multiprocessing import Process,Queue
import queue

def f(q2):
    q2.put("hkshhdjskajdksa")

if __name__=="__main__":
    q=Queue()
    p=Process(target=f,args=(q,))
    p.start()
    print(q.get())
    p.join()

'''manager是用来传递对象'''
from multiprocessing import Process,Manager
import os

def f(d,l,l1):
    d[os.getpid()] = os.getpid()   #os.getpid():Return the current process id.
    l.append(os.getpid())
    l1.append(os.getpid())  #l1属于直接传递，不能回传
    # print(l)
    # print("l1:***",l1)   #普通列表在子进程中每次会获得一个新值，但都会被下一个值覆盖
    # print(d)

if __name__ == "__main__":
    with Manager() as mager:
        d = mager.dict()         #由manager生成的字典
        l = mager.list(range(5)) #由manager生成的列表
        l1 = []                  #普通列表无法共享数据，最后仍旧是空列表
        p_list = []
        for i in range(10):
            p = Process(target=f,args=(d,l,l1))
            p.start()
            p.join()
        print("l:",l)
        print(l1)
        print("d:",d)

数据传递：

'''主进程(父类进程) 子进程'''
'''管道通信实现数据之间的传递'''
# from multiprocessing import Process,Pipe
#
# def f(con):
#     con.send("hello from child1")
#     con.send("hello from child2")
#     print("parent news:",con.recv())
#     con.close()
#
# if __name__ == "__main__":
#     Parent_con,Child_con = Pipe()
#     p = Process(target=f,args=(Child_con,))
#     p.start()
#     print(Parent_con.recv())
#     print(Parent_con.recv())
#     Parent_con.send("from parent")
#     p.join()



'''运行结果'''
hello from child1
hello from child2
parent news: from parent

 进程锁（Lock）：

屏幕存在共享，多进程可以同时使用屏幕，进程加锁的目的在于，确保屏幕被独个进程使用。

from multiprocessing import Process,Lock

def f(l,i):
    l.acquire()
    print("+++",i)
    l.release()

if __name__ == "__main__":
    lock = Lock()   #加锁的目的是为了确保屏幕被单独占用
    for num in range(100):
        Process(target=f,args=(lock,num)).start()


'''运行结果'''
+++ 0    
+++ 1
+++ 2
+++ 3
+++ 4
+++ 5
.
.
.(不在这里演示完所有内容)

进程池（pool）：

python中，进程池内部会维护一个进程序列。当需要时，程序会去进程池中获取一个进程。

如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

from multiprocessing import Process,Pool
import time
import os

def foo(i):
    time.sleep(2)
    print("in the process:",os.getpid())
    return i+100


def bar(args):
    print("system done",args)

if __name__ == "__main__":
    pool = Pool(5)
    for i in range(10):
        # pool.apply(func=foo,args=(i,))
        #生成进程，把pool放入容器
        #apply本身是一个串行方法,不受join影响

        pool.apply_async(func=foo,args=(i,),callback=bar)
        #apply_async是一个并行方法,受join影响
        #callback()回调函数为主进程操作，进程池一旦开始运行，回调函数会自动执行


    print("end")
    pool.close()  #pool关闭是需要时间的，所以在close之后再join
    pool.join()

pool的内置方法：

• apply    串行方法。从进程池里取一个进程并同步执行，不受join影响
• apply_async    并行方法。从进程池里取出一个进程并异步执行，受join影响
• terminate    立刻关闭进程池
• join 主进程等待所有子进程执行完毕，必须在close或terminete之后（如上述代码）
• close 等待所有进程结束才关闭线程池

#
# Module providing the `Pool` class for managing a process pool
#模块提供用于管理进程池的“池”类
# multiprocessing/pool.py
#
# Copyright (c) 2006-2008, R Oudkerk
# Licensed to PSF under a Contributor Agreement.
#

__all__ = ['Pool', 'ThreadPool']

#
# Imports
#

import threading
import queue
import itertools
import collections
import os
import time
import traceback

# If threading is available then ThreadPool should be provided.  Therefore
# we avoid top-level imports which are liable to fail on some systems.
from . import util
from . import get_context, TimeoutError

#
# Constants representing the state of a pool
#表示池状态的常数

RUN = 0
CLOSE = 1
TERMINATE = 2

#
# Miscellaneous
#

job_counter = itertools.count()

def mapstar(args):
    return list(map(*args))

def starmapstar(args):
    return list(itertools.starmap(args[0], args[1]))

#
# Hack to embed stringification of remote traceback in local traceback
#

class RemoteTraceback(Exception):
    def __init__(self, tb):
        self.tb = tb
    def __str__(self):
        return self.tb

class ExceptionWithTraceback:
    def __init__(self, exc, tb):
        tb = traceback.format_exception(type(exc), exc, tb)
        tb = ''.join(tb)
        self.exc = exc
        self.tb = '
"""
%s"""' % tb
    def __reduce__(self):
        return rebuild_exc, (self.exc, self.tb)

def rebuild_exc(exc, tb):
    exc.__cause__ = RemoteTraceback(tb)
    return exc

#
# Code run by worker processes
#

class MaybeEncodingError(Exception):
    """Wraps possible unpickleable errors, so they can be
    safely sent through the socket."""
    #包装可能出现的无法拾取的错误，以便通过套接字安全地发送这些错误。

    def __init__(self, exc, value):
        self.exc = repr(exc)
        self.value = repr(value)
        super(MaybeEncodingError, self).__init__(self.exc, self.value)

    def __str__(self):
        return "Error sending result: '%s'. Reason: '%s'" % (self.value,
                                                             self.exc)

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self)


def worker(inqueue, outqueue, initializer=None, initargs=(), maxtasks=None,
           wrap_exception=False):
    if (maxtasks is not None) and not (isinstance(maxtasks, int)
                                       and maxtasks >= 1):
        raise AssertionError("Maxtasks {!r} is not valid".format(maxtasks))
    put = outqueue.put
    get = inqueue.get
    if hasattr(inqueue, '_writer'):
        inqueue._writer.close()
        outqueue._reader.close()

    if initializer is not None:
        initializer(*initargs)

    completed = 0
    while maxtasks is None or (maxtasks and completed < maxtasks):
        try:
            task = get()
        except (EOFError, OSError):
            util.debug('worker got EOFError or OSError -- exiting')
            break

        if task is None:
            util.debug('worker got sentinel -- exiting')
            break

        job, i, func, args, kwds = task
        try:
            result = (True, func(*args, **kwds))
        except Exception as e:
            if wrap_exception and func is not _helper_reraises_exception:
                e = ExceptionWithTraceback(e, e.__traceback__)
            result = (False, e)
        try:
            put((job, i, result))
        except Exception as e:
            wrapped = MaybeEncodingError(e, result[1])
            util.debug("Possible encoding error while sending result: %s" % (
                wrapped))
            put((job, i, (False, wrapped)))

        task = job = result = func = args = kwds = None
        completed += 1
    util.debug('worker exiting after %d tasks' % completed)

def _helper_reraises_exception(ex):
    'Pickle-able helper function for use by _guarded_task_generation.'
    #用于_guarded_task_generation的可选择助手函数。
    raise ex

#
# Class representing a process pool  类表示进程池
#

class Pool(object):
    '''
    Class which supports an async version of applying functions to arguments.
    '''
    #类，该类支持将函数应用于参数的异步版本。
    _wrap_exception = True

    def Process(self, *args, **kwds):
        return self._ctx.Process(*args, **kwds)

    def __init__(self, processes=None, initializer=None, initargs=(),
                 maxtasksperchild=None, context=None):
        self._ctx = context or get_context()
        self._setup_queues()
        self._taskqueue = queue.SimpleQueue()
        self._cache = {}
        self._state = RUN
        self._maxtasksperchild = maxtasksperchild
        self._initializer = initializer
        self._initargs = initargs

        if processes is None:
            processes = os.cpu_count() or 1
        if processes < 1:
            raise ValueError("Number of processes must be at least 1")

        if initializer is not None and not callable(initializer):
            raise TypeError('initializer must be a callable')

        self._processes = processes
        self._pool = []
        self._repopulate_pool()

        self._worker_handler = threading.Thread(
            target=Pool._handle_workers,
            args=(self, )
            )
        self._worker_handler.daemon = True
        self._worker_handler._state = RUN
        self._worker_handler.start()


        self._task_handler = threading.Thread(
            target=Pool._handle_tasks,
            args=(self._taskqueue, self._quick_put, self._outqueue,
                  self._pool, self._cache)
            )
        self._task_handler.daemon = True
        self._task_handler._state = RUN
        self._task_handler.start()

        self._result_handler = threading.Thread(
            target=Pool._handle_results,
            args=(self._outqueue, self._quick_get, self._cache)
            )
        self._result_handler.daemon = True
        self._result_handler._state = RUN
        self._result_handler.start()

        self._terminate = util.Finalize(
            self, self._terminate_pool,
            args=(self._taskqueue, self._inqueue, self._outqueue, self._pool,
                  self._worker_handler, self._task_handler,
                  self._result_handler, self._cache),
            exitpriority=15
            )

    def _join_exited_workers(self):
        """Cleanup after any worker processes which have exited due to reaching
        their specified lifetime.  Returns True if any workers were cleaned up.
        """
        #在由于达到指定的生存期而退出的任何工作进程之后进行清理。如果有工人被清理干净，
        # 返回True。
        cleaned = False
        for i in reversed(range(len(self._pool))):
            worker = self._pool[i]
            if worker.exitcode is not None:
                # worker exited
                util.debug('cleaning up worker %d' % i)
                worker.join()
                cleaned = True
                del self._pool[i]
        return cleaned

    def _repopulate_pool(self):
        """Bring the number of pool processes up to the specified number,
        for use after reaping workers which have exited.
        """
        #将池进程的数量增加到指定的数量，以便在收割已退出的工人后使用。
        for i in range(self._processes - len(self._pool)):
            w = self.Process(target=worker,
                             args=(self._inqueue, self._outqueue,
                                   self._initializer,
                                   self._initargs, self._maxtasksperchild,
                                   self._wrap_exception)
                            )
            self._pool.append(w)
            w.name = w.name.replace('Process', 'PoolWorker')
            w.daemon = True
            w.start()
            util.debug('added worker')

    def _maintain_pool(self):
        """Clean up any exited workers and start replacements for them.
        """
        #清理所有离职的员工，并开始替换他们。
        if self._join_exited_workers():
            self._repopulate_pool()

    def _setup_queues(self):
        self._inqueue = self._ctx.SimpleQueue()
        self._outqueue = self._ctx.SimpleQueue()
        self._quick_put = self._inqueue._writer.send
        self._quick_get = self._outqueue._reader.recv

    def apply(self, func, args=(), kwds={}):
        '''
        Equivalent of `func(*args, **kwds)`.
        Pool must be running.
        '''
        #相当于“func(*args， ** kwds)”。池必须正在运行。
        return self.apply_async(func, args, kwds).get()

    def map(self, func, iterable, chunksize=None):
        '''
        Apply `func` to each element in `iterable`, collecting the results
        in a list that is returned.
        '''
        #对“iterable”中的每个元素应用“func”，在返回的列表中收集结果。
        return self._map_async(func, iterable, mapstar, chunksize).get()

    def starmap(self, func, iterable, chunksize=None):
        '''
        Like `map()` method but the elements of the `iterable` are expected to
        be iterables as well and will be unpacked as arguments. Hence
        `func` and (a, b) becomes func(a, b).
        '''
        #方法类似于“map()”，但“iterable”的元素也应该是可迭代的，并将作为参数解压缩。
        # 因此“func”和(a, b)变成了func(a, b)。
        return self._map_async(func, iterable, starmapstar, chunksize).get()

    def starmap_async(self, func, iterable, chunksize=None, callback=None,
            error_callback=None):
        '''
        Asynchronous version of `starmap()` method.
        '''
        #异步版本的“starmap()”方法。
        return self._map_async(func, iterable, starmapstar, chunksize,
                               callback, error_callback)

    def _guarded_task_generation(self, result_job, func, iterable):
        '''Provides a generator of tasks for imap and imap_unordered with
        appropriate handling for iterables which throw exceptions during
        iteration.'''
        #为imap和imap_unordered提供任务生成器，并为迭代期间抛出异常的迭代提供适当的处理。
        try:
            i = -1
            for i, x in enumerate(iterable):
                yield (result_job, i, func, (x,), {})
        except Exception as e:
            yield (result_job, i+1, _helper_reraises_exception, (e,), {})

    def imap(self, func, iterable, chunksize=1):
        '''
        Equivalent of `map()` -- can be MUCH slower than `Pool.map()`.
        '''
        #等价于“map()”——可能比“Pool.map()”慢得多。
        if self._state != RUN:
            raise ValueError("Pool not running")
        if chunksize == 1:
            result = IMapIterator(self._cache)
            self._taskqueue.put(
                (
                    self._guarded_task_generation(result._job, func, iterable),
                    result._set_length
                ))
            return result
        else:
            if chunksize < 1:
                raise ValueError(
                    "Chunksize must be 1+, not {0:n}".format(
                        chunksize))
            task_batches = Pool._get_tasks(func, iterable, chunksize)
            result = IMapIterator(self._cache)
            self._taskqueue.put(
                (
                    self._guarded_task_generation(result._job,
                                                  mapstar,
                                                  task_batches),
                    result._set_length
                ))
            return (item for chunk in result for item in chunk)

    def imap_unordered(self, func, iterable, chunksize=1):
        '''
        Like `imap()` method but ordering of results is arbitrary.
        '''
        #Like `imap()`方法，但结果的顺序是任意的。
        if self._state != RUN:
            raise ValueError("Pool not running")
        if chunksize == 1:
            result = IMapUnorderedIterator(self._cache)
            self._taskqueue.put(
                (
                    self._guarded_task_generation(result._job, func, iterable),
                    result._set_length
                ))
            return result
        else:
            if chunksize < 1:
                raise ValueError(
                    "Chunksize must be 1+, not {0!r}".format(chunksize))
            task_batches = Pool._get_tasks(func, iterable, chunksize)
            result = IMapUnorderedIterator(self._cache)
            self._taskqueue.put(
                (
                    self._guarded_task_generation(result._job,
                                                  mapstar,
                                                  task_batches),
                    result._set_length
                ))
            return (item for chunk in result for item in chunk)

    def apply_async(self, func, args=(), kwds={}, callback=None,
            error_callback=None):
        '''
        Asynchronous version of `apply()` method.  “apply()”方法的异步版本。
        '''
        if self._state != RUN:
            raise ValueError("Pool not running")
        result = ApplyResult(self._cache, callback, error_callback)
        self._taskqueue.put(([(result._job, 0, func, args, kwds)], None))
        return result

    def map_async(self, func, iterable, chunksize=None, callback=None,
            error_callback=None):
        '''
        Asynchronous version of `map()` method. 方法的异步版本
        '''
        return self._map_async(func, iterable, mapstar, chunksize, callback,
            error_callback)

    def _map_async(self, func, iterable, mapper, chunksize=None, callback=None,
            error_callback=None):
        '''
        Helper function to implement map, starmap and their async counterparts.
        '''
        #帮助函数实现映射，星图和他们的异步对等。
        if self._state != RUN:
            raise ValueError("Pool not running")
        if not hasattr(iterable, '__len__'):
            iterable = list(iterable)

        if chunksize is None:
            chunksize, extra = divmod(len(iterable), len(self._pool) * 4)
            if extra:
                chunksize += 1
        if len(iterable) == 0:
            chunksize = 0

        task_batches = Pool._get_tasks(func, iterable, chunksize)
        result = MapResult(self._cache, chunksize, len(iterable), callback,
                           error_callback=error_callback)
        self._taskqueue.put(
            (
                self._guarded_task_generation(result._job,
                                              mapper,
                                              task_batches),
                None
            )
        )
        return result

    @staticmethod
    def _handle_workers(pool):
        thread = threading.current_thread()

        # Keep maintaining workers until the cache gets drained, unless the pool
        # is terminated.
        #继续维护worker，直到缓存耗尽，除非池终止。
        while thread._state == RUN or (pool._cache and thread._state != TERMINATE):
            pool._maintain_pool()
            time.sleep(0.1)
        # send sentinel to stop workers
        pool._taskqueue.put(None)
        util.debug('worker handler exiting')

    @staticmethod
    def _handle_tasks(taskqueue, put, outqueue, pool, cache):
        thread = threading.current_thread()

        for taskseq, set_length in iter(taskqueue.get, None):
            task = None
            try:
                # iterating taskseq cannot fail
                #迭代taskseq不会失败
                for task in taskseq:
                    if thread._state:
                        util.debug('task handler found thread._state != RUN')
                        break
                    try:
                        put(task)
                    except Exception as e:
                        job, idx = task[:2]
                        try:
                            cache[job]._set(idx, (False, e))
                        except KeyError:
                            pass
                else:
                    if set_length:
                        util.debug('doing set_length()')
                        idx = task[1] if task else -1
                        set_length(idx + 1)
                    continue
                break
            finally:
                task = taskseq = job = None
        else:
            util.debug('task handler got sentinel')

        try:
            # tell result handler to finish when cache is empty
            #告诉结果处理程序在缓存为空时结束
            util.debug('task handler sending sentinel to result handler')
            outqueue.put(None)

            # tell workers there is no more work
            util.debug('task handler sending sentinel to workers')
            for p in pool:
                put(None)
        except OSError:
            util.debug('task handler got OSError when sending sentinels')

        util.debug('task handler exiting')

    @staticmethod
    def _handle_results(outqueue, get, cache):
        thread = threading.current_thread()

        while 1:
            try:
                task = get()
            except (OSError, EOFError):
                util.debug('result handler got EOFError/OSError -- exiting')
                return

            if thread._state:
                assert thread._state == TERMINATE, "Thread not in TERMINATE"
                util.debug('result handler found thread._state=TERMINATE')
                break

            if task is None:
                util.debug('result handler got sentinel')
                break

            job, i, obj = task
            try:
                cache[job]._set(i, obj)
            except KeyError:
                pass
            task = job = obj = None

        while cache and thread._state != TERMINATE:
            try:
                task = get()
            except (OSError, EOFError):
                util.debug('result handler got EOFError/OSError -- exiting')
                return

            if task is None:
                util.debug('result handler ignoring extra sentinel')
                continue
            job, i, obj = task
            try:
                cache[job]._set(i, obj)
            except KeyError:
                pass
            task = job = obj = None

        if hasattr(outqueue, '_reader'):
            util.debug('ensuring that outqueue is not full')
            # If we don't make room available in outqueue then
            # attempts to add the sentinel (None) to outqueue may
            # block.  There is guaranteed to be no more than 2 sentinels.
            #如果我们不在outqueue中留出可用的空间，那么尝试将sentinel (None)
            # 添加到outqueue可能会阻塞。保证不超过2个哨兵。
            try:
                for i in range(10):
                    if not outqueue._reader.poll():
                        break
                    get()
            except (OSError, EOFError):
                pass

        util.debug('result handler exiting: len(cache)=%s, thread._state=%s',
              len(cache), thread._state)

    @staticmethod
    def _get_tasks(func, it, size):
        it = iter(it)
        while 1:
            x = tuple(itertools.islice(it, size))
            if not x:
                return
            yield (func, x)

    def __reduce__(self):
        raise NotImplementedError(
              'pool objects cannot be passed between processes or pickled'
            #不能在进程之间传递池对象或pickle池对象
              )

    def close(self):
        util.debug('closing pool')
        if self._state == RUN:
            self._state = CLOSE
            self._worker_handler._state = CLOSE

    def terminate(self):
        util.debug('terminating pool')
        self._state = TERMINATE
        self._worker_handler._state = TERMINATE
        self._terminate()

    def join(self):
        util.debug('joining pool')
        if self._state == RUN:
            raise ValueError("Pool is still running")
        elif self._state not in (CLOSE, TERMINATE):
            raise ValueError("In unknown state")
        self._worker_handler.join()
        self._task_handler.join()
        self._result_handler.join()
        for p in self._pool:
            p.join()

    @staticmethod
    def _help_stuff_finish(inqueue, task_handler, size):
        # task_handler may be blocked trying to put items on inqueue
        #试图将项放入inqueue时可能阻塞task_handler
        util.debug('removing tasks from inqueue until task handler finished')
        inqueue._rlock.acquire()
        while task_handler.is_alive() and inqueue._reader.poll():
            inqueue._reader.recv()
            time.sleep(0)

    @classmethod
    def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool,
                        worker_handler, task_handler, result_handler, cache):
        # this is guaranteed to only be called once 这保证只调用一次
        util.debug('finalizing pool')

        worker_handler._state = TERMINATE
        task_handler._state = TERMINATE

        util.debug('helping task handler/workers to finish')
        cls._help_stuff_finish(inqueue, task_handler, len(pool))

        if (not result_handler.is_alive()) and (len(cache) != 0):
            raise AssertionError(
                "Cannot have cache with result_hander not alive")

        result_handler._state = TERMINATE
        outqueue.put(None)                  # sentinel

        # We must wait for the worker handler to exit before terminating
        # workers because we don't want workers to be restarted behind our back.
        #我们必须在终止工人之前等待工人处理程序退出，因为我们不希望工人在我们背后重新启动。
        util.debug('joining worker handler')
        if threading.current_thread() is not worker_handler:
            worker_handler.join()

        # Terminate workers which haven't already finished.
        if pool and hasattr(pool[0], 'terminate'):
            util.debug('terminating workers')
            for p in pool:
                if p.exitcode is None:
                    p.terminate()

        util.debug('joining task handler')
        if threading.current_thread() is not task_handler:
            task_handler.join()

        util.debug('joining result handler')
        if threading.current_thread() is not result_handler:
            result_handler.join()

        if pool and hasattr(pool[0], 'terminate'):
            util.debug('joining pool workers')
            for p in pool:
                if p.is_alive():
                    # worker has not yet exited
                    util.debug('cleaning up worker %d' % p.pid)
                    p.join()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.terminate()

#
# Class whose instances are returned by `Pool.apply_async()`
#

class ApplyResult(object):

    def __init__(self, cache, callback, error_callback):
        self._event = threading.Event()
        self._job = next(job_counter)
        self._cache = cache
        self._callback = callback
        self._error_callback = error_callback
        cache[self._job] = self

    def ready(self):
        return self._event.is_set()

    def successful(self):
        if not self.ready():
            raise ValueError("{0!r} not ready".format(self))
        return self._success

    def wait(self, timeout=None):
        self._event.wait(timeout)

    def get(self, timeout=None):
        self.wait(timeout)
        if not self.ready():
            raise TimeoutError
        if self._success:
            return self._value
        else:
            raise self._value

    def _set(self, i, obj):
        self._success, self._value = obj
        if self._callback and self._success:
            self._callback(self._value)
        if self._error_callback and not self._success:
            self._error_callback(self._value)
        self._event.set()
        del self._cache[self._job]

AsyncResult = ApplyResult       # create alias -- see #17805

#
# Class whose instances are returned by `Pool.map_async()`
#

class MapResult(ApplyResult):

    def __init__(self, cache, chunksize, length, callback, error_callback):
        ApplyResult.__init__(self, cache, callback,
                             error_callback=error_callback)
        self._success = True
        self._value = [None] * length
        self._chunksize = chunksize
        if chunksize <= 0:
            self._number_left = 0
            self._event.set()
            del cache[self._job]
        else:
            self._number_left = length//chunksize + bool(length % chunksize)

    def _set(self, i, success_result):
        self._number_left -= 1
        success, result = success_result
        if success and self._success:
            self._value[i*self._chunksize:(i+1)*self._chunksize] = result
            if self._number_left == 0:
                if self._callback:
                    self._callback(self._value)
                del self._cache[self._job]
                self._event.set()
        else:
            if not success and self._success:
                # only store first exception
                self._success = False
                self._value = result
            if self._number_left == 0:
                # only consider the result ready once all jobs are done
                if self._error_callback:
                    self._error_callback(self._value)
                del self._cache[self._job]
                self._event.set()

#
# Class whose instances are returned by `Pool.imap()`
#

class IMapIterator(object):

    def __init__(self, cache):
        self._cond = threading.Condition(threading.Lock())
        self._job = next(job_counter)
        self._cache = cache
        self._items = collections.deque()
        self._index = 0
        self._length = None
        self._unsorted = {}
        cache[self._job] = self

    def __iter__(self):
        return self

    def next(self, timeout=None):
        with self._cond:
            try:
                item = self._items.popleft()
            except IndexError:
                if self._index == self._length:
                    raise StopIteration from None
                self._cond.wait(timeout)
                try:
                    item = self._items.popleft()
                except IndexError:
                    if self._index == self._length:
                        raise StopIteration from None
                    raise TimeoutError from None

        success, value = item
        if success:
            return value
        raise value

    __next__ = next                    # XXX

    def _set(self, i, obj):
        with self._cond:
            if self._index == i:
                self._items.append(obj)
                self._index += 1
                while self._index in self._unsorted:
                    obj = self._unsorted.pop(self._index)
                    self._items.append(obj)
                    self._index += 1
                self._cond.notify()
            else:
                self._unsorted[i] = obj

            if self._index == self._length:
                del self._cache[self._job]

    def _set_length(self, length):
        with self._cond:
            self._length = length
            if self._index == self._length:
                self._cond.notify()
                del self._cache[self._job]

#
# Class whose instances are returned by `Pool.imap_unordered()`
#类，其实例由' Pool.imap_unordered() '返回

class IMapUnorderedIterator(IMapIterator):

    def _set(self, i, obj):
        with self._cond:
            self._items.append(obj)
            self._index += 1
            self._cond.notify()
            if self._index == self._length:
                del self._cache[self._job]

#
#
#

class ThreadPool(Pool):
    _wrap_exception = False

    @staticmethod
    def Process(*args, **kwds):
        from .dummy import Process
        return Process(*args, **kwds)

    def __init__(self, processes=None, initializer=None, initargs=()):
        Pool.__init__(self, processes, initializer, initargs)

    def _setup_queues(self):
        self._inqueue = queue.SimpleQueue()
        self._outqueue = queue.SimpleQueue()
        self._quick_put = self._inqueue.put
        self._quick_get = self._outqueue.get

    @staticmethod
    def _help_stuff_finish(inqueue, task_handler, size):
        # drain inqueue, and put sentinels at its head to make workers finish
        #排干队伍内的水，并在其头部放置哨兵，使工人完成工作
        try:
            while True:
                inqueue.get(block=False)
        except queue.Empty:
            pass
        for i in range(size):
            inqueue.put(None)