完全搞懂Python的线程机制

import threading
import time

exitFlag = 0

# 继承的方法写一个多线程逻辑
class myThread (threading.Thread):

    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter
        
    def run(self):
        print ("开始线程：" + self.name)
        print_time(self.name, self.counter, 5)
        print ("退出线程：" + self.name)

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            threadName.exit()
        # 这里模拟一个阻塞
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()

# join等待
thread1.join()
thread2.join()
print ("退出主线程")

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.

    """
    # 线程是否初始化（这里就是刚那个init函数
    if not self._initialized:
        raise RuntimeError("thread.__init__() not called")

    # 设置线程的开始状态（把线程状态置为start
    if self._started.is_set():
        raise RuntimeError("threads can only be started once")

    # 加锁，调用GIL
    with _active_limbo_lock:
        _limbo[self] = self

    # 去启动新的线程
    try:
        _start_new_thread(self._bootstrap, ())
    # 如果启动出了问题
    except Exception:
        # 把锁给del了
        with _active_limbo_lock:
            del _limbo[self]
        raise
    # 释放锁，阻塞。直到被唤醒/超时
    self._started.wait()

void PyEval_InitThreads(void)
{
    if (gil_created())
        return;
    // 创建GIL锁
    create_gil();
    // 申请GIL锁
    take_gil(PyThreadState_GET());
    // 主线程
    main_thread = PyThread_get_thread_ident();
    // 如果没有等待锁
    if (!pending_lock)
        // 创建等待锁
        pending_lock = PyThread_allocate_lock();
}

static void take_gil(PyThreadState *tstate)
{
    MUTEX_LOCK(gil_mutex);    // 加锁

    if (!_Py_atomic_load_relaxed(&gil_locked))
        // 获取 GIL，若已释放，直接获取，跳转
        goto _ready;

    while (_Py_atomic_load_relaxed(&gil_locked)) {
        // GIL 未释放
        int timed_out = 0;
        unsigned long saved_switchnum;
        // 记录切换次数
        saved_switchnum = gil_switch_number;
        // 利用 pthread_cond_tim 阻塞等待超时
        COND_TIMED_WAIT(gil_cond, gil_mutex, INTERVAL, timed_out);
        /* 等待超时，仍未释放, 发送释放请求信号 */
        if (timed_out &&
            _Py_atomic_load_relaxed(&gil_locked) &&
            gil_switch_number == saved_switchnum) {
            // 设置 gil_drop_request=1，eval_breaker=1
            // 释放信号
            SET_GIL_DROP_REQUEST();
        }
    }
_ready:
    /* We now hold the GIL */
    _Py_atomic_store_relaxed(&gil_locked, 1); // 设置 GIL 占用
    _Py_ANNOTATE_RWLOCK_ACQUIRED(&gil_locked, /*is_write=*/1);

    if (tstate != (PyThreadState*)_Py_atomic_load_relaxed(&gil_last_holder)) {
        _Py_atomic_store_relaxed(&gil_last_holder, (uintptr_t)tstate);
        ++gil_switch_number;
    }

    if (_Py_atomic_load_relaxed(&gil_drop_request)) {
        // 重置 gil_drop_request=0
        RESET_GIL_DROP_REQUEST();
    }
    if (tstate->async_exc != NULL) {
        _PyEval_SignalAsyncExc();
    }

    MUTEX_UNLOCK(gil_mutex);    // 解锁
}

PyObject *
_PyEval_EvalFrameDefault(PyFrameObject *f, int throwflag){
    ...
    for (;;) {
        if (_Py_atomic_load_relaxed(&eval_breaker)) {
            if (_Py_OPCODE(*next_instr) == SETUP_FINALLY ||
                    _Py_OPCODE(*next_instr) == YIELD_FROM) {
                goto fast_next_opcode;
            }
            if (_Py_atomic_load_relaxed(&pendingcalls_to_do)) {
                if (Py_MakePendingCalls() < 0)
                    goto error;
            }

            // 如果检测到gil_drop_request，释放GIL
            if (_Py_atomic_load_relaxed(&gil_drop_request)) {
                /* Give another thread a chance */
                if (PyThreadState_Swap(NULL) != tstate)
                    Py_FatalError("ceval: tstate mix-up");
                drop_gil(tstate);

                // 继续去调用GIL抢占
                take_gil(tstate);

                // 检查是否退出线程
                if (_Py_Finalizing && _Py_Finalizing != tstate) {
                    drop_gil(tstate);
                    PyThread_exit_thread();
                }

                if (PyThreadState_Swap(tstate) != NULL)
                    Py_FatalError("ceval: orphan tstate");
            }
        }
        ...
    }
}

import threading
import time

exitFlag = 0

# 利用继承的方法
class myThread (threading.Thread):

    def __init__(self, threadID, name, counter):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.counter = counter

    def run(self):
        print ("开始线程：" + self.name)
        print_time(self.name, self.counter, 5)
        print ("退出线程：" + self.name)

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()

thread1.join()
thread2.join()
print ("退出主线程")

import threading
import time
from queue import Queue


queue = Queue(20)


# 生产者
def Producer():
    i = 0
    print("开始线程")
    while True:
        i = i + 1
        print("生产数据", i, "现有数据", queue.qsize())
        time.sleep(1)
        queue.put(i)

# 消费者
def Consumer():
    while True:
        i = queue.get()
        time.sleep(0.5)
        print("消费数据", i)


if __name__ == "__main__":
    Th1 = threading.Thread(target=Producer, )

    Th2 = threading.Thread(target=Consumer, )

    Th2.start()
    Th1.start()

    Th1.join()
    Th2.join()

import threading
import time

count = 0


# 做一个累加的函数
def add_num():
    global count
    if lock.acquire():  # 获得锁，并返回True
        tmp = count
        time.sleep(0.001)
        count = tmp + 1
        lock.release()  # 执行完释放锁


def run(add_fun):
    global count
    thread_list = []

    # 累加100次
    for i in range(100):
        t = threading.Thread(target=add_fun)
        t.start()
        thread_list.append(t)  

    # 等待线程执行完
    for j in thread_list:
        j.join()

    print(count)


if __name__ == '__main__':
    # 添加全局锁
    lock = threading.Lock()
    # 执行函数
    run(add_num)

from socket import AF_INET, SOCK_STREAM, socket
from concurrent.futures import ThreadPoolExecutor


# 假装跑一个服务器client
def echo_client(sock, client_addr):
    '''
    Handle a client connection
    '''
    print('Got connection from', client_addr)
    while True:
        msg = sock.recv(65536)
        if not msg:
            break
        sock.sendall(msg)
    print('Client closed connection')
    sock.close()


# 假装跑一个server
def echo_server(addr):
    # 定义一个线程池
    pool = ThreadPoolExecutor(128)
    sock = socket(AF_INET, SOCK_STREAM)
    sock.bind(addr)
    sock.listen(5)
    while True:
        client_sock, client_addr = sock.accept()
        # 从池子里拿线程消费
        pool.submit(echo_client, client_sock, client_addr)

echo_server(('',15000))