python 多进程队列 写文件速度问题

tjlizheng

生产者进程 消费者进程   通过队列保存中间结果  在消费者进程里面 取队列数据 进行写文件 速度一直起不来  2M/s


在消费者进程里面 不取数据 直接循环写 速度也起不来 ，不知道是不是python 队列锁机制的问题

不知道大牛们有没有好的解决办法

timespace

除了Python的queue，有用其它的进程同步机制吗？还有写入频率和数据块大小？

laputa73

用gevent的queue

q1208c

直接单线程写一下试试先

timespace

还是从自己代码找问题吧，Queue的效率不至于那么低。

CPU：单核3.4GHz
内存：1GB
OS：CentOS 6.4
Python: 2.6.6
场景：在竞争条件下queue的效率，多个生产者进程（示例中为10个）写queue，每个记录为4KiB，每个生产者写100K个记录，一个消费者进程读queue，尽量消除IO的影响，使消费者写/dev/null
结论：平均数据写入速度高于50MiB/s

代码：

1. #!/usr/bin/env python
   
2. 
   
3. import io, os, sys
   
4. from multiprocessing import Process, Queue
   
5. from datetime import datetime
   
6. 
   
7. def producer(q, no):
   
8.     blk_data = bytearray(4096) # bytes
   
9.     cnt = 0
   
10.     while cnt < 10 ** 5:
    
11.         q.put(blk_data)
    
12.         cnt += 1
    
13. 
    
14. def consumer(q):
    
15.     begin_time = datetime.now()
    
16.     blk_cnt = 0
    
17.     with io.open(os.devnull, 'wb') as fp:
    
18.         while True:
    
19.             blk_data = q.get()
    
20.             if not isinstance(blk_data, bytearray):
    
21.                 break
    
22.             fp.write(blk_data)
    
23.             blk_cnt += 1
    
24.     end_time = datetime.now()
    
25.     print 'duration', end_time - begin_time
    
26.     print 'data size', blk_cnt / 256 , 'MiB'
    
27. 
    
28. def main():
    
29.     if len(sys.argv) != 2:
    
30.         sys.exit('{0} producers'.format(sys.argv[0]))
    
31. 
    
32.     log_queue = Queue() # the queue of log record
    
33.     proc_producer = [
    
34.             Process(target=producer, args=(log_queue, i))
    
35.             for i in range(int(sys.argv[1], 10))]
    
36.     proc_consumer = Process(target=consumer, args=(log_queue,))
    
37.     for p in proc_producer:
    
38.         p.start()
    
39.     proc_consumer.start()
    
40. 
    
41.     for p in proc_producer:
    
42.         p.join()
    
43.     log_queue.put(None) # notify consumer to exit
    
44.     proc_consumer.join()
    
45. 
    
46. if __name__ == '__main__':
    
47.     main()
    

复制代码

运行：

1. $ ./mp_queue.py 10
   
2. duration 0:01:07.505177
   
3. data size 3906 MiB
   

复制代码

tjlizheng

blk_data = bytearray(4096)
如果是字符串  只有 10个字节左右  不用字节数组  速度就降下来了

是不是要把字符串连接起来存呢？  

icymirror

回复 6# tjlizheng


    这里速度的关键是 bytearray 的大小和 IO 的速度。

    比如：你把 timespace 代码的 4096 改成 32，当然，对应的 size 计算的部分也从 /256 改成 /32768，就会发现两个运行的传输效果大约前者是后者的：16 倍 < 4096 / 32 = 128
    也就是说：在 buffersize (就是 bytearray )空间比较大的时候，时间主要是在花在传输和切换两方面上。 当 buffersize 比较小时，主要时间是在切换上。
    从比值可以猜测，当 buffersize 在 <256 的时候，效率可能是个正比与 buffersize的值，这时主要时间用在切换线程。

    基于 timespace 的代码实验的结果是： buffersize = 256 传输 244M 需要 22s => 244 / 22 ~ 11, buffersize = 32, 传输 30M 需要 20s => 30 / 20 ~ 1.5。速度比和 buffersize 的比值都接近 8，应当是切换影响大。

    但是：当 buffersize 继续上上升时，读取数据也会花时间:
    我实验的结果是：buffersize = 1024 传输 976M 需要 40s => 976 / 40 ~ 24.4, buffersize = 4096, 传输 3906M 需要 70s => 3906 / 70 ~ 56。
    速度比和 buffersize 比已经不是很一致了， buffersize 比是 4 倍， 但是速度比是 2.3 倍左右。

    所以，当数据少的时候，最好提升一次性传输的数据大小，可以在近乎相同的切换次数的情况下提升传输数据量，当数据多的时候，buffersize 仍然有效，但是 I/O 影响也开始上升，提升就没有那么明显了。

    就你的数据传输问题来说，因为字符串太小，建议先把一次性传输的内容数量提升，如果你的字符串只有 10 几个字节，最好把至少 256/10 大约 20 个左右的字符串一起传送会好些。