python 实现线程之间的通信示例

时间：2020-08-15 python教程查看: 851

前言：因为GIL的限制，python的线程是无法真正意义上并行的。相对于异步编程，其性能可以说不是一个等量级的。为什么我们还要学习多线程编程呢，虽然说异步编程好处多，但编程也较为复杂，逻辑不容易理解，学习成本和维护成本都比较高。毕竟我们大部分人还是适应同步编码的，除非一些需要高性能处理的地方采用异步。

首先普及下进程和线程的概念：

进程：进程是操作系统资源分配的基本单位。

线程：线程是任务调度和执行的基本单位。

一个应用程序至少一个进程，一个进程至少一个线程。

两者区别：同一进程内的线程共享本进程的资源如内存、I/O、cpu等，但是进程之间的资源是独立的。

一、多线程

python 可以通过 thread 或 threading 模块实现多线程，threading 相比 thread 提供了更高阶、更全面的线程管理。我们下文主要以 threading 模块介绍多线程的基本用法。

import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)

 def run(self):
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 线程个数
 threads.append(thread(str(a)))
for t in threads: # 开启线程
 t.start()
for t in threads: # 阻塞线程
 t.join()
print('END')

输出：
线程3:Now timestamp is 1557386184.7574518
线程2:Now timestamp is 1557386184.7574518
线程0:Now timestamp is 1557386184.7574518
线程1:Now timestamp is 1557386184.7574518
线程4:Now timestamp is 1557386184.7582724
END

start() 方法开启子线程。运行多次 start() 方法代表开启多个子线程。

join() 方法用来阻塞主线程，等待子线程执行完成。举个例子，主线程A创建了子线程B，并使用了 join() 方法，主线程A在 join() 处就被阻塞了，等待子线程B完成后，主线程A才能执行 print('END')。如果没有使用 join() 方法，主线程A创建子线程B后，不会等待子线程B，直接执行 print('END')，如下：

import threading
import time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)

 def run(self):
 time.sleep(1)
 print('%s:Now timestamp is %s'%(self.name,time.time()))

threads = []
for a in range(int(5)): # 线程个数
 threads.append(thread(str(a)))
for t in threads: # 开启线程
 t.start()
# for t in threads: # 阻塞线程
# t.join()
print('END')

输出：
END
线程0:Now timestamp is 1557386321.376941
线程3:Now timestamp is 1557386321.377937
线程1:Now timestamp is 1557386321.377937
线程2:Now timestamp is 1557386321.377937
线程4:Now timestamp is 1557386321.377937

二、线程之间的通信

1.threading.Lock()

如果多个线程对某一资源同时进行修改，可能会存在不可预知的情况。为了修改数据的正确性，需要把这个资源锁住，只允许线程依次排队进去获取这个资源。当线程A操作完后，释放锁，线程B才能进入。如下脚本是开启多个线程修改变量的值，但输出结果每次都不一样。

import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  Order(self.threadname)

t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

接下来我们用 threading.Lock() 锁住这个变量，等操作完再释放这个锁。lock.acquire() 给资源加一把锁，对资源处理完成之后，lock.release() 再释放锁。以下脚本执行结果都是一样的，但速度会变慢，因为线程只能一个个的通过。

import threading

money = 0
def Order(n):
 global money
 money = money + n
 money = money - n

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 for i in range(1000000):
  lock.acquire()
  Order(self.threadname)
  lock.release()
# print('%s:Now timestamp is %s'%(self.name,time.time()))

lock = threading.Lock()
t1 = thread('1')
t2 = thread('5')
t1.start()
t2.start()
t1.join()
t2.join()
print(money)

2.threading.Rlock()

用法和 threading Lock() 一致，区别是 threading.Rlock() 允许多次锁资源，acquire() 和 release() 必须成对出现，也就是说加了几把锁就得释放几把锁。

lock = threading.Lock()
# 死锁
lock.acquire()
lock.acquire()
print('...')
lock.release()
lock.release()

rlock = threading.RLock()
# 同一线程内不会阻塞线程
rlock.acquire()
rlock.acquire()
print('...')
rlock.release()
rlock.release()

3.threading.Condition()

threading.Condition() 可以理解为更加高级的锁，比 Lock 和 Rlock 的用法更高级，能处理一些复杂的线程同步问题。threading.Condition() 创建一把资源锁（默认是Rlock），提供 acquire() 和 release() 方法，用法和 Rlock 一致。此外 Condition 还提供 wait()、Notify() 和 NotifyAll() 方法。

wait()：线程挂起，直到收到一个 Notify() 通知或者超时（可选参数），wait() 必须在线程得到 Rlock 后才能使用。

Notify() ：在线程挂起的时候，发送一个通知，让 wait() 等待线程继续运行，Notify() 也必须在线程得到 Rlock 后才能使用。 Notify(n=1)，最多唤醒 n 个线程。

NotifyAll() ：在线程挂起的时候，发送通知，让所有 wait() 阻塞的线程都继续运行。

举例说明下 Condition() 使用

import threading,time

def TestA():
 cond.acquire()
 print('李白：看见一个敌人，请求支援')
 cond.wait()
 print('李白：好的')
 cond.notify()
 cond.release()

def TestB():
 time.sleep(2)
 cond.acquire()
 print('亚瑟：等我...')
 cond.notify()
 cond.wait()
 print('亚瑟：我到了，发起冲锋...')

if __name__=='__main__':
 cond = threading.Condition()
 testA = threading.Thread(target=TestA)
 testB = threading.Thread(target=TestB)
 testA.start()
 testB.start()
 testA.join()
 testB.join()

输出
李白：看见一个敌人，请求支援
亚瑟：等我...
李白：好的
亚瑟：我到了，发起冲锋...

4.threading.Event()

threading.Event() 原理是在线程中立了一个 Flag ，默认值是 False ，当一个或多个线程遇到 event.wait() 方法时阻塞，直到 Flag 值变为 True 。threading.Event() 通常用来实现线程之间的通信，使一个线程等待其他线程的通知，把 Event 传递到线程对象中。

event.wait() ：阻塞线程，直到 Flag 值变为 True

event.set() ：设置 Flag 值为 True

event.clear() ：修改 Flag 值为 False

event.isSet() : 仅当 Flag 值为 True 时返回

下面这个例子，主线程启动子线程后 sleap 2秒，子线程因为 event.wait() 被阻塞。当主线程醒来后执行 event.set() ，子线程才继续运行，两者输出时间差 2s。

import threading
import datetime,time

class thread(threading.Thread):
 def __init__(self, threadname):
 threading.Thread.__init__(self, name='线程' + threadname)
 self.threadname = int(threadname)

 def run(self):
 event.wait()
 print('子线程运行时间：%s'%datetime.datetime.now())

if __name__ == '__main__':
 event = threading.Event()
 t1 = thread('0')
 #启动子线程
 t1.start()
 print('主线程运行时间：%s'%datetime.datetime.now())
 time.sleep(2)
 # Flag设置成True
 event.set()
 t1.join()

输出
主线程运行时间：2019-05-30 15:51:49.690872
子线程运行时间：2019-05-30 15:51:51.691523

5.其他方法

threading.active_count()：返回当前存活的线程对象的数量

threading.current_thread()：返回当前线程对象

threading.enumerate()：返回当前所有线程对象的列表

threading.get_ident()：返回线程pid

threading.main_thread()：返回主线程对象

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持python博客。

展开全文