在Python中多个线程写入同一CSV

Question

我是Python中的多线程新手，目前正在编写一个附加到csv文件的脚本。如果我要将多个线程提交到一个concurrent.futures.ThreadPoolExecutor中，该concurrent.futures.ThreadPoolExecutor将行追加到csv文件中。如果追加是这些线程执行的唯一与文件相关的操作，我可以做些什么来保证线程安全？

我的代码的简化版本：

with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    for count,ad_id in enumerate(advertisers):

        downloadFutures.append(executor.submit(downloadThread, arguments.....))
        time.sleep(random.randint(1,3)) 

我的线程类是：

def downloadThread(arguments......):

                #Some code.....

                writer.writerow(re.split(',', line.decode()))

我应该设置一个单独的单线程执行器来处理写操作，还是担心我只是附加？

编辑：我应该详细说明，当写操作发生时，在下一次附加文件之间的时间间隔可能会有很大的变化，我只是担心在测试我的脚本时没有发生这种情况，我更希望对此进行讨论。

Answer 1

下面是一些代码，它还处理引起unicode问题的头痛：

def ensure_bytes(s):
    return s.encode('utf-8') if isinstance(s, unicode) else s

class ThreadSafeWriter(object):
'''
>>> from StringIO import StringIO
>>> f = StringIO()
>>> wtr = ThreadSafeWriter(f)
>>> wtr.writerow(['a', 'b'])
>>> f.getvalue() == "a,b\\r\\n"
True
'''

def __init__(self, *args, **kwargs):
    self._writer = csv.writer(*args, **kwargs)
    self._lock = threading.Lock()

def _encode(self, row):
    return [ensure_bytes(cell) for cell in row]

def writerow(self, row):
    row = self._encode(row)
    with self._lock:
        return self._writer.writerow(row)

def writerows(self, rows):
    rows = (self._encode(row) for row in rows)
    with self._lock:
        return self._writer.writerows(rows)

# example:
with open('some.csv', 'w') as f:
    writer = ThreadSafeWriter(f)
    writer.write([u'中文', 'bar'])

更详细的解决方案是here