python-threading

• python-threading
  • • threading
    • Thread
    • 创建线程
    • Thread方法属性
    • 守护线程
    • 线程锁Lock
      • threading.Lock
      • threading.RLock
    • 事件对象Event
    • Condition
    • Timer

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

线程也叫轻量级进程，是操作系统能够进行运算调度的最小单位，它被包涵在进程之中，是进程中的实际运作单位，进程中线程之间的隔离程度要小，它们共享内存、文件句柄和其他进程应有的状态。一个线程可以创建和撤销另一个线程，。一条线程指的是进程中一个单一顺序的控制流，一个进程中可以并发多个线程，每条线程并行执行不同的任务。

threading

Thread

线程对象

from threading import Thread

Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

- group	   none，在实现ThreadGroup类时保留用于将来的扩展。
- target   是run()方法调用的可调用对象。默认为None，意味着什么都没有被调用。
- name     是线程名称。默认情况下，唯一名称由“Thread- N ” 形式构成，其中N是小十进制数。
- args 	   是目标调用的参数元组。默认为()
- kwargs   调用目标函数的关键字参数，默认是{}
- daemon   当为True时，MainThread 结束，子线程也立马结束

Thread 的生命周期
创建对象时，代表 Thread 内部被初始化
创建好一个线程对象后，该对象并不会立即执行,调用 start() 方法后，thread 会开始运行
thread 代码正常运行结束或者是遇到异常,线程会终止

创建线程

# 方式1函数方法
import threading
import time
def run(n):
    print('task',n)
    time.sleep(1)
    print('2s')
    time.sleep(1)
    print('1s')
    time.sleep(1)

# target是要执行的函数名（不是函数），args是函数对应的参数，以元组的形式存在
t1 = threading.Thread(target=run,args=('t1',))    
t2 = threading.Thread(target=run,args=('t2',))
t1.start()
t2.start()  
    
# 方法2 类继承
# 直接初始化一个 Thread，然后，现在还有一种方式就是自定义一个 Thread 的子类，然后复写它的 run() 方法
class MyThread(threading.Thread):
    def __init__(self,n):
        super(MyThread,self).__init__()   #重构run函数必须写
        self.n = n

    def run(self):
        print('task',self.n)
        time.sleep(1)
        print('2s')
        time.sleep(1)
        print('1s')
        time.sleep(1)
    
t1 = MyThread('t1')
t2 = MyThread('t2')
t1.start()
t2.start()

Thread方法属性

start()   开始线程活动，只能调用一次
run()     线程活动的方式， 可以在子类型中重载方法
join()    等待。指导线程终结，阻塞条用这个方法的线程，可以使用多次

# 两个线程是同时运行的，但如果让一个先运行，一个后运行
# 调用一个 Thread 的 join() 方法，可以阻塞自身所在的线程

import threading
import time

def test():
    for i in range(5):
        print(threading.current_thread().name+' test ',i)
        time.sleep(0.5)

thread = threading.Thread(target=test,name='TestThread')
thread.start()
thread.join()  # 阻塞,等待thread优先执行完成

for i in range(5):
    print(threading.current_thread().name+' main ', i)
    print(thread.name+' is alive ', thread.isAlive())
    time.sleep(1)
# 主线程创建了 TestThread 对象后start,然后通过调用 join() 方法,指定 thread 线程优先执行完毕 实现等待
# 结果
TestThread test  0
TestThread test  1
TestThread test  2
TestThread test  3
TestThread test  4
MainThread main  0
TestThread is alive  False
MainThread main  1
TestThread is alive  False
MainThread main  2
TestThread is alive  False
MainThread main  3
TestThread is alive  False
MainThread main  4
TestThread is alive  False


# join()的作用 优先执行  
# 用来设计 守护线程 r
def run():
    time.sleep(2)
    print('当前线程 -', threading.current_thread().name)
    time.sleep(2)

if __name__ == '__main__':

    start_time = time.time()

    print('这是主线程：', threading.current_thread().name)
    thread_list = []
    for i in range(5):
        t = threading.Thread(target=run)
        thread_list.append(t)

    for t in thread_list:
        t.setDaemon(True)  # 把子进程设置为守护线程，必须在start()之前设置
        t.start()

    for t in thread_list:
        t.join()

    print('主线程结束了！' , threading.current_thread().name)
    print('一共用时：', time.time()-start_time)
    
# 结果
这是主线程： MainThread
当前线程 - Thread-4
当前线程 - Thread-3
当前线程 - Thread-1
当前线程 - Thread-2
当前线程 - Thread-5
主线程结束了！ MainThread
一共用时： 4.02971315383911

import threading
import time

def run(n):
    print("task", n)
    time.sleep(1)       #此时子线程停1s
    print('3')
    time.sleep(1)
    print('2')
    time.sleep(1)
    print('1')

if __name__ == '__main__':
    t = threading.Thread(target=run, args=("t1",))
    t.start()
    t.join(1.0) # 设置主线程等待子线程结束
    print("end")

----------------------------------

>>> task t1
>>> 3
>>> 2
>>> 1
>>> end

守护线程

守护线程，又称后台线程，它是在后台运行的，如果所有前台线程都死亡，那么后台线程就会自动死亡。

非守护线程：一般创建的线程默认就是非守护线程，包括主线程也是，即在Python程序退出时，如果还有非守护线程在运行，程序会等待直到所有非守护线程都结束后才会退出。

import threading
import time

def run(n):
    print("task", n)
    time.sleep(1)       #此时子线程停1s
    print('3')
    time.sleep(1)
    print('2')
    time.sleep(1)
    print('1')

if __name__ == '__main__':
    t = threading.Thread(target=run, args=("t1",))
    t.setDaemon(True)   #把子进程设置为守护线程，必须在start()之前设置,前台线程结束,则后台线程随之结束
    t.start()
    print("end")   # 此处主线程结束，子线程也随之结束。
    
----------------------------------

>>> task t1
>>> end

线程锁Lock

当有多个线程，且它们同时访问同一资源时，需要考虑如何避免线程冲突。解决办法是使用线程锁 , 线程锁的代价是，失去异步效果

原始锁处于 "锁定" 或者 "非锁定" 两种状态之一。它被创建时为非锁定状态。它有两个基本方法， acquire()和 release()

threading.Lock

实现原始锁对象的类。一旦一个线程获得一个锁，会阻塞随后尝试获得锁的线程，直到它被释放；任何线程都可以释放它。

acquire(blocking=True, timeout=-1)：获取锁，并将锁的状态改为“锁定”,成功返回True，失败返回False。当一个线程获得锁时，会阻塞其他尝试获取锁的线程,直到这个锁被释放掉。即将无限阻塞等待直到获得锁
release()：释放一个锁，并将其状态改为“非锁定”,任何线程都可以释放锁

import time
import threading

# 创建锁
lock = threading.Lock()
global_resource = [None] * 5

def change_resource(para, sleep):
    lock.acquire()
    # 如果不加锁，第一个线程 global_resource中是乱的 ['hello', 'hi', 'hi', 'hello', 'hello']
    # 第二个线程运行结束后，global_resource中还是乱的 ['hello', 'hi', 'hi', 'hi', 'hi']
    global global_resource
    for i in range(len(global_resource)):
        global_resource[i] = para
        time.sleep(sleep)
    print("修改全局变量为：", global_resource)
    lock.release()
def main():
    thread_hi = threading.Thread(target=change_resource, args=('hi', 2))
    thread_hello = threading.Thread(target=change_resource, args=('hello', 1))
    thread_hi.start()
    thread_hello.start()

if __name__ == '__main__':
    main()
    

threading.RLock

递归锁：RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次acquire。直到一个线程所有的acquire都被release，其他的线程才能获得资源。

使用普通锁时，对于一些可能造成死锁的情况，可以考虑使用递归锁来解决。

所谓的死锁就是指由多个线程直接，各自持有某些资源，又在申请其他线程所持有的资源，各自坚持着都不释放资源，一直坚持着，这就是死锁。

1、互斥条件：当一个进程在访问一个资源的时候，其他进程只能等待。即任何时候一个资源只能给一个进程使用。
2、不可剥夺条件:一个进程在访问一个资源时，其他进程只能等该进程使用完释放资源，不可强行剥夺。
3、请求和保持条件：当一个进程在申请它所需的资源时，并不会释放已有的资源

'''
死锁的例子
'''
from threading import Thread,Lock,RLock
mutexA=Lock()
mutexB=Lock()

class mythread(Thread):
    def run(self):
        self.func1()
        self.func2()

    def func1(self):
        mutexA.acquire()
        print("线程:{name} 拿到的了A锁".format(name=self.name))
        mutexB.acquire()
        print("线程:{name}拿到了B锁".format(name=self.name))
        mutexB.release()
        print("线程:{name}释放了B锁".format(name=self.name))
        mutexA.release()
        print("线程:{name}释放了A锁".format(name=self.name))

    def func2(self):
        mutexB.acquire()
        print("线程：{name} 拿到了B锁".format(name=self.name))
        mutexA.acquire()
        print("线程:{name} 拿到了A锁".format(name=self.name))
        mutexA.release()
        print("线程:{name} 释放了A锁".format(name=self.name))
        mutexB.release()
        print("线程:{name} 释放了B锁".format(name=self.name))

if __name__ == '__main__':
    for i in range(3):
        t=mythread()
        t.start()

所谓的递归锁就是指一个线程可以多次申请同一把锁，但是不会造成死锁。这就可以用来解决上面的死锁问题

'''
使用递归锁避免产生死锁
'''
from threading import Thread,Lock,RLock
mutexA=mutexB=RLock()

class mythread(Thread):
    def run(self):
        self.func1()
        self.func2()

    def func1(self):
        mutexA.acquire()
        print("线程:{name} 拿到的了A锁".format(name=self.name))
        mutexB.acquire()
        print("线程:{name}拿到了B锁".format(name=self.name))
        mutexB.release()
        print("线程:{name}释放了B锁".format(name=self.name))
        mutexA.release()
        print("线程:{name}释放了A锁".format(name=self.name))

    def func2(self):
        mutexB.acquire()
        print("线程：{name} 拿到了B锁".format(name=self.name))
        mutexA.acquire()
        print("线程:{name} 拿到了A锁".format(name=self.name))
        mutexA.release()
        print("线程:{name} 释放了A锁".format(name=self.name))
        mutexB.release()
        print("线程:{name} 释放了B锁".format(name=self.name))

if __name__ == '__main__':
    for i in range(10):
        t=mythread()
        t.start()

事件对象Event

信号事件， 主要作用是用于线程间的通信，确切说用于主线性控制其他线程的执行，实现线程同步

线程的一个关键特性是每个线程都是独立运行且状态不可预测,为了通过判断线程的状态来确定下一步的操作，使用 Event 来实现线程同步的问题。

通过threading.Event()可以创建一个事件管理标志，该标志event默认为False，event对象主要有四种方法可以调用

• event.wait(timeout=None)：调用该方法的线程会被阻塞，如果设置了timeout参数，超时后，线程会停止阻塞继续执行
• event.set()：将event的标志设置为True，此时所有等待中的线程将被唤醒，调用wait()方法的线程将不会被阻塞;
• event.clear()：将event的标志设置为False，所有调用wait()方法的线程将被阻塞，
• event.isSet()：判断event的标志是否为True返回True。

# https://www.cnblogs.com/guyuyun/p/11185832.html
import time
import threading

# 创建事件对象，内部标志默认为False
event = threading.Event()

def student_exam(student_id):
    print('学生%s等监考老师发卷。。。' % student_id)
    event.wait()
    print('开始考试了！')


def invigilate_teacher():
    time.sleep(5)
    print('考试时间到，学生们可以开始考试了！')
    # 设置内部标志为True，并唤醒所有等待的线程
    event.set()


def main():
    for student_id in range(3):
        threading.Thread(target=student_exam, args=(student_id, )).start()

    threading.Thread(target=invigilate_teacher).start()


if __name__ == '__main__':
    main()
    
学生0等监考老师发卷。。。
学生1等监考老师发卷。。。
学生2等监考老师发卷。。。
考试时间到，学生们可以开始考试了！
开始考试了！
开始考试了！
开始考试了！

# 来自 https://www.cnblogs.com/rongye/p/9978261.html
# 一个好玩的程序
def lighter():
    count = 0
    event.set()  # 先设置绿灯
    while True:
        if count > 5 and count < 10:  # 改成红灯
            event.clear()  # 把标志位清了
            print("\033[41;1mred light is on....\033[0m")
        elif count > 10:
            event.set()  # 变绿灯
            count = 0
        else:
            print("\033[42;1mgreen light is on....\033[0m")
        time.sleep(1)
        count += 1

def car(name):
    while True:
        if event.is_set():  # 代表绿灯
            print("[%s] running..." % name)
            time.sleep(1)
        else:
            print("[%s] sees red light , waiting...." % name)
            event.wait()
            print("\033[34;1m[%s] green light is on, start going...\033[0m" % name)

event = Event()
light = threading.Thread(target=lighter, )
light.start()
car1 = threading.Thread(target=car, args=("beike"))
car1.start()

Condition

threading.Condition(lock=None)一个条件变量对象允许一个或多个线程等待，直到被另一个线程通知。lock参数必须是一个Lock对象或者RLock对象，并且会作为底层锁使用

• acquire(*args)： 请求底层锁。此方法调用底层锁对应的方法和返回对应方法的返回值。
• **release()： ** 释放底层锁。此方法调用底层所对应的方法，没有返回值。
• **wait(timeout=None)： ** 释放锁，等待直到被通知（再获取锁）或者发生超时事件。
• wait_for(predicate, timeout=None)：与wait方法相似，等待，直到条件计算为True
• notify(n=1)：唤醒一个等待这个条件的线程
• notify_all()：唤醒所有等待这个条件的线程。

"""
让一个线程等待，直到另一个线程通知
"""
import time
import threading


# 创建条件变量对象
condition_lock = threading.Condition()

PRE = 0

# predicate可调用函数
def pre():
    print(PRE)
    return PRE

def test_thread_hi():
    # 在使用wait/wait_for之前必须先获得锁
    condition_lock.acquire()

    print('等待线程test_thread_hello的通知')
    # 先执行一次pre，返回False后释放掉锁，等另一个线程释放掉锁后再次执行pre，返回True后再次获取锁
    # wait_for的返回值不是True和False，而是predicate参数的返回值
    condition_lock.wait_for(pre)   # 等待，直到条件计算为True
    # condition_lock.wait()
    print('继续执行')

    # 不要忘记使用wait/wait_for之后要释放锁
    condition_lock.release()

def test_thread_hello():
    time.sleep(2)
    condition_lock.acquire()

    global PRE
    PRE = 1
    print('修改PRE值为1')

    print('通知线程test_thread_hi可以准备获取锁了')
    condition_lock.notify()
    
    # 先notify/notify_all之后在释放锁
    condition_lock.release()
    print('你获取锁吧')

def main():
    thread_hi = threading.Thread(target=test_thread_hi)
    thread_hello = threading.Thread(target=test_thread_hello)
    thread_hi.start()
    thread_hello.start()


if __name__ == '__main__':
    main()

等待线程test_thread_hello的通知
0
修改PRE值为1
通知线程test_thread_hi可以准备获取锁了
你获取锁吧
1
继续执行

Timer

表示一个操作应该在等待一定的时间之后运行 --- 相当于一个定时器

def hello():
    print("hello, world")

t = Timer(30.0, hello)
t.start()  # after 30 seconds, "hello, world" will be printed

详解Python中的Thread线程模块 (jquerycn.cn)