我有一个多线程程序，我在其中创建了一个生成器函数，然后将它传递给新线程。我希望它本质上是共享/全局的，因此每个线程都可以从生成器中获取下一个值。
使用这样的生成器是否安全，或者我会遇到从多个线程访问共享生成器的问题/条件？
如果没有，有没有更好的方法来解决这个问题？我需要一些可以循环遍历列表并为调用它的线程生成下一个值的东西。
【问题讨论】：
标签： python multithreading thread-safety generator
感谢 IIRC python freenode，这里是 python 3.x 的工作解决方案
默认情况下生成器不是线程安全的，但这里是如何使它们成为线程安全的
def my_generator():
while True:
for x in range(10):
yield x
class LockedIterator(object):
def __init__(self, it):
self._lock = threading.Lock()
self._it = iter(it)
def __iter__(self):
return self
def __next__(self):
with self._lock:
return next(self._it)
n = LockedIterator(my_generator)
next(n)
next(n)
next(n)
或者使用函数
def threadsafe_iter(iterable):
lock = threading.Lock()
iterator = iter(iterable)
while True:
with lock:
for value in iterator:
break
else:
return
yield value
n = threadsafe_iter(my_generator)
next(n)
next(n)
next(n)
【讨论】：
生成器对象本身是线程安全的，就像任何受 GIL 保护的 PyObject 一样。但是线程试图从已经在其他线程中处于执行状态的生成器中获取下一个元素（在yield之间执行生成器代码）会得到ValueError：
ValueError: generator already executing
示例代码：
from threading import Thread
from time import sleep
def gen():
sleep(1)
yield
g = gen()
Thread(target=g.__next__).start()
Thread(target=g.__next__).start()
结果：
Exception in thread Thread-2:
Traceback (most recent call last):
File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
self.run()
File "/usr/lib/python3.8/threading.py", line 870, in run
self._target(*self._args, **self._kwargs)
ValueError: generator already executing
但是，实际上这根本与线程无关。并且可以在单个线程中复制：
def gen():
yield next(g)
g = gen()
next(g)
【讨论】：
已编辑以在下面添加基准。
你可以用锁包裹一个生成器。例如，
import threading
class LockedIterator(object):
def __init__(self, it):
self.lock = threading.Lock()
self.it = it.__iter__()
def __iter__(self): return self
def next(self):
self.lock.acquire()
try:
return self.it.next()
finally:
self.lock.release()
gen = [x*2 for x in [1,2,3,4]]
g2 = LockedIterator(gen)
print list(g2)
在我的系统上锁定需要 50 毫秒，队列需要 350 毫秒。当您确实有队列时，队列很有用；例如，如果您有传入的 HTTP 请求，并且您希望将它们排队以供工作线程处理。 （这不适合 Python 迭代器模型——一旦迭代器用完项目，它就完成了。）如果你真的有一个迭代器，那么 LockedIterator 是一种更快、更简单的方法来使其线程安全。
from datetime import datetime
import threading
num_worker_threads = 4
class LockedIterator(object):
def __init__(self, it):
self.lock = threading.Lock()
self.it = it.__iter__()
def __iter__(self): return self
def next(self):
self.lock.acquire()
try:
return self.it.next()
finally:
self.lock.release()
def test_locked(it):
it = LockedIterator(it)
def worker():
try:
for i in it:
pass
except Exception, e:
print e
raise
threads = []
for i in range(num_worker_threads):
t = threading.Thread(target=worker)
threads.append(t)
t.start()
for t in threads:
t.join()
def test_queue(it):
from Queue import Queue
def worker():
try:
while True:
item = q.get()
q.task_done()
except Exception, e:
print e
raise
q = Queue()
for i in range(num_worker_threads):
t = threading.Thread(target=worker)
t.setDaemon(True)
t.start()
t1 = datetime.now()
for item in it:
q.put(item)
q.join()
start_time = datetime.now()
it = [x*2 for x in range(1,10000)]
test_locked(it)
#test_queue(it)
end_time = datetime.now()
took = end_time-start_time
print "took %.01f" % ((took.seconds + took.microseconds/1000000.0)*1000)
【讨论】：
这取决于您使用的 python 实现。在 CPython 中，GIL 使得对 python 对象的所有操作都是线程安全的，因为在任何给定时间只有一个线程可以执行代码。
【讨论】：
不，它们不是线程安全的。您可以在以下位置找到有关生成器和多线程的有趣信息：
【讨论】：
它不是线程安全的；同时调用可能会交错，并与局部变量混淆。
常见的做法是使用主从模式（现在在 PC 中称为农民工模式）。创建第三个生成数据的线程，并在主从之间添加一个队列，从属将从队列中读取，主将向其写入。标准队列模块提供必要的线程安全，并安排阻塞主节点，直到从节点准备好读取更多数据。
【讨论】：