在处理事件队列的过程中不少情况是采用轮询的方式进行的。
例如如下例子所示。在Example中，主线程和子线程通过队列的形式进行通信，为模拟业务，主线程将随时获取到的待处理任务放入对应的任务队列(将一随机数放入随机的队列中，之后等待随机的时间)，子线程发现有队列中有待处理的事件就将其取出进行处理(取出队列中的数字进行打印)。
import time
from collections import deque
from random import Random
import threading
class Example():
def __init__(self):
self.r = Random(time.time())
def worker_run(self, name, q):
while True:
if q:
print("%s pop number: %s, %s numbers left" % (name, q.pop(), len(q)))
time.sleep(0.001)
def manager_run(self,worker_num):
queue_list = [deque() for i in range(worker_num)]
worker_list = [threading.Thread(target=self.worker_run,
args=["Thread-%s" % i, queue_list[i]]) for i in range(worker_num)]
for worker in worker_list:
worker.start()
while True:
i = 0
while i< self.r.randint(0,20):
queue_list[self.r.randint(0,worker_num - 1)].appendleft(
self.r.randint(1,100)
)
i += 1
time.sleep(self.r.random()/100)
if __name__ == '__main__':
Example().manager_run(3)
进行功能分离简单，目的性明确。但是存在效率性的问题，如果想提高触发效率，那么需要将worker_run中的每轮sleep时间设置的非常小，甚至是0。这么就造成了另一个问题，CPU的损耗问题。无意义的while死循环无疑会造成CPU的无意义损耗。这个对机器性能消耗非常巨大的。所以，需要对机制进行修改，由轮询转为事件驱动。这样的更改无疑会对代码整体结构有较大的调整。但是是否有简单的方式可以快速达到目的，且更代码改较量小。
不难想到，只要在事件队列为空时暂停工作线程，队列放入值时重启工作线程即可。原本，直接对线程进行中断和继续是最好的解决方案，但是在python中threading.Thread并不支持。在官方文档如是说：
The design of this module is loosely based on Java’s threading model. However, where Java makes locks and condition variables basic behavior of every object, they are separate objects in Python. Python’s Thread
class supports a subset of the behavior of Java’s Thread class; currently, there are no priorities, no thread groups, and threads cannot be destroyed, stopped, suspended, resumed, or interrupted. The static methods of Java’s Thread class, when implemented, are mapped to module-level functions.
这意味着，没有一种方法直接对子线程进行暂停、重启等。所以换了一个思路，使用threading.Event实现信号量作为暂停重启的替代方案。当需要暂停时，使用threading.Event.wait()等待信号。当需要继续时，使用threading.Event.set()发出信号。这样就解决了线程暂停、继续的问题。
另一个问题是如何在队列发生改变时触发信号量的更改，不同的变量可以通过不同的方式实现，事实上每一种变量都用不同的解决方案。
- 对于内部实例化的示例可以通过继承重新构建同名class的方式，重写修改内容的方法实现触发信号量。
- 外部变量则可以通过methodType的方式绑定重写对于的方法。
- 实例的基础类型属性则可以通过@property的方式在setter中加入信号触发。
- ...
所以可以对上述示例进行如下修改。
import time
from collections import deque as dq
from random import Random
import threading
#重写用于事件队列的deque，使得对deque的变动会触发信号量
class deque(dq):
#存储信号的集合
sign = set()
#重写append、appendleft、extend、extendleft方法
def append(self, *args, **kwargs):
dq.append(self, *args, **kwargs)
self.sign_set()
def appendleft(self, *args, **kwargs):
dq.appendleft(self, *args, **kwargs)
self.sign_set()
def extend(self, *args, **kwargs):
dq.extend(self, *args, **kwargs)
def extendleft(self, *args, **kwargs):
dq.extendleft(self, *args, **kwargs)
#触发信号
def sign_set(self):
for s in self.sign:
s.set()
#添加一个信号
def add_sign(self,s):
self.sign.add(s)
class Example():
def __init__(self):
self.r = Random(time.time())
#添加一个信号量绑定给deque，当发现deque中没有值之后等待信号传入
def worker_run(self, name, q):
sign = threading.Event()
q.add_sign(sign)
while True:
if q:
print("%s pop number: %s, %s numbers left" % (name, q.pop(), len(q)))
else:
sign.wait()
sign.clear()
def manager_run(self,worker_num):
queue_list = [deque() for i in range(worker_num)]
worker_list = [threading.Thread(target=self.worker_run,
args=["Thread-%s" % i, queue_list[i]])
for i in range(worker_num)]
for worker in worker_list:
worker.start()
while True:
i = 0
while i< self.r.randint(0,20):
queue_list[self.r.randint(0,worker_num - 1)].appendleft(
self.r.randint(1,100)
)
i += 1
time.sleep(self.r.random())
if __name__ == '__main__':
Example().manager_run(3)
通过这种方式最小限度的更改代码，几乎无代码结构变动。就可以将轮询触发变为事件驱动，在不影响原有业务的前提下降低系统的无意义负载。
网友评论