资源有限的 Python 多处理 问题说明问题：我的尝试

如何解决资源有限的 Python 多处理 问题说明问题：我的尝试

问题说明

我目前正在构建一个包含三个任务的交换抓取器，每个任务都独立运行process：

• #1：接收实时网络馈送：非常快的数据传入，立即放入 multiprocessing 队列并继续。
• #2：使用队列数据并优化：使用我编写的一些逻辑来使用和优化它。慢但不是太慢，最终在数据进入缓慢时赶上并清除队列。
• #3：使用 bz2 压缩提要并上传到我的 s3 存储桶：每小时，我都会压缩优化的数据（以进一步减小文件大小），然后上传到我的 s3桶。这在我的机器上大约需要 10-20 秒。

我遇到的问题是这些任务中的每一个都需要自己的并行 process。生产者 (#1) 不能进行优化 (#2)，否则它会停止馈送连接并且网站会杀死我的套接字，因为线程 #1 没有响应。上传器 (#3) 不能与任务 #2 在同一进程上运行，否则我会填满队列，我永远赶不上。我试过这个：不起作用。

这个刮刀在我的本地机器上工作得很好，每个任务都有自己的进程。但是我真的不想在一个3核机器上花很多钱，当它部署在服务器上时。我发现 Digital Ocean 的 4vCPU 选项最便宜，每月 40 美元。但我想知道是否有比支付 4 核更好的方法。

需要注意的一些事情：在我的基础 16" MBP 上，任务 #1 使用 99% 的 CPU，任务 #2 使用 20-30% 的 CPU，任务 #3 休眠到小时，所以它主要使用 0.5 -1% CPU。

问题：

• 如果我在一台 2 核机器上运行三个进程，这与运行两个进程是否有效相同？我知道它 depends on system scheduling，但这是否意味着它会在压缩时停止，或者一直移动直到压缩结束？启动（并支付）一个每小时仅使用一次的全新核心似乎真的很浪费。但是这个每小时的任务让整个队列停顿太多，我不知道如何解决这个问题。

• 在我在同一个进程/核心上压缩我的文件时，我是否可以继续任务#2？

• 如果我运行一个 bash 脚本来进行压缩，那仍然会使软件停止吗？我的电脑是 6 核，所以我不能真正在本地测试服务器的约束

• 是否有比 DigitalOcean 更便宜的替代品？老实说，我对 AWS 感到害怕，因为我听说过人们因意外使用而收到 1,000 美元账单的恐怖故事。我更喜欢像 DigitalOcean 这样更可预测的东西

我的尝试

如前所述，我尝试在同一进程中组合 Task#2 和 Task#3。一旦压缩开始，它就会停止。压缩是同步的，并使用 the code from this thread 完成。找不到异步 bz2 压缩，但我不确定这是否有助于不拖延 Task#2。

---

PS：我真的试图避免带着这样的悬而未决的问题来到 StackOverflow，因为我知道这些问题得到了不好的反馈，但替代方案是尝试并投入大量时间+金钱说实话，我对云计算了解不多。我更喜欢一些专家的意见

解决方法

要点#1：

您将遇到的所有操作系统都使用抢占式调度来在进程之间切换。这应该保证每个进程在任何远程现代硬件上至少每秒恢复几次（只要进程使用 cpu，而不是等待像文件或套接字 io 这样的中断）。基本上，在 2 核 CPU 上运行数百个进程根本不是问题。如果总负载太多，一切都会运行得更慢，但不会完全停顿。

要点#2：

多线程？您可能会发现压缩/存储受到更多的 IO 限制，因此这里的线程可能没问题。由于子线程可以完全访问父线程的内存空间，因此您甚至可能会看到减少进程之间传输数据的开销（取决于您当前的执行方式）带来的好处。

要点#3：

shell 脚本只是另一个进程，所以回答 #1 并没有太大区别。但是，请对此进行测试，因为 python bzip 很可能比 shell bzip 慢得多（取决于您提供数据的方式，以及它试图将数据放在哪里）...

要点 #4：

绝对不是 S.O. 的合适问题

我的建议：

分析您的代码... 使摄取过程尽可能高效，并在进程之间发送尽可能少的数据。仅从套接字读取数据并将其发送以进行处理的进程应该占用最少的 CPU。默认的 multiprocessing.Queue 不是非常有效，因为它腌制数据，通过管道发送，然后在另一端取消腌制。如果您的数据可以分成固定大小的块，请考虑使用几个 multiprocessing.shared_memory.SharedMemory 缓冲区在它们之间进行交换。分块数据流还可以更轻松地并行化数据消费阶段，从而更好地利用您拥有的任何 CPU 资源。

编辑：通过共享内存发送数据块的伪代码示例

import multiprocessing as mp
from contextlib import contextmanager
from collections import namedtuple
from ctypes import c_int8
import socket
import time

STALE_DATA = 0 #data waiting to be overwritten
NEW_DATA = 1 #data waiting to be processed

def producer_func(buffers):
    shm_objects = {}
    for buffer in buffers:
        shm_objects[buffer.shm_name] = mp.shared_memory.SharedMemory(name=buffer.shm_name,create=False)
        #buffer.state.value = 0 #value was initialized as stale at creation (data waiting to be overwritten)
    with socket.create_connection(...) as s: #however you're reading data
        while True: #for each chunk of data
            while True: #until we get an open buffer
                for buffer in buffers: #check each buffer object
                    #if buffer isn't being processed right now,and data has already been processed
                    if buffer.lock.acquire(False):
                        if buffer.state.value==STALE_DATA: 
                            shm = shm_objects[buffer.shm_name]
                            break #break out of two loops
                        else:
                            buffer.lock.release()
                else:
                    continue
                break
            
            s.recv_into(shm.buf) #put the data in the buffer
            buffer.state.value = NEW_DATA #flag the data as new
            buffer.lock.release() #release the buffer to be processed
    #when you receive some sort of shutdown signal:
    for shm in shm_objects:
        shm.close()

def consumer_func(buffers):
    shm_objects = {}
    for buffer in buffers:
        shm_objects[buffer.shm_name] = mp.shared_memory.SharedMemory(name=buffer.shm_name,create=False)
        #buffer.state.value = 0 #value was initialized as stale at creation (data waiting to be overwritten)
    while True: #for each chunk of data
        while True: #until we get a buffer of data waiting to be processed
            for buffer in buffers:
                #if buffer isn't being processed right now,and data hasn't already been processed
                if buffer.lock.acquire(False):
                    if buffer.state.value==NEW_DATA:
                        shm = shm_objects[buffer.shm_name]
                        break #break out of two loops
                    else:
                        buffer.lock.release()
            else:
                continue
            break
        process_the_data(shm.buf) #do your data reduction here
        buffer.state.value = STALE_DATA
        buffer.lock.release()
    #when you receive some sort of shutdown signal:
    for shm in shm_objects:
        shm.close()
        
Buffer = namedtuple("Buffer",['shm_name','lock','state'])

if __name__ == "__main__":
    n_buffers = 4 # 4 buffers to swap between
    #each buffer should be bigger than you will ever expect a message to be.
    #using larger chunks is better for overhead (don't be processing chunks of less than a couple Kib at a time)
    shm_objects = [mp.shared_memory.SharedMemory(create=True,size=2**20) for _ in range(n_buffers)] # 1MiB buffers
    buffers = [Buffer(shm.name,mp.Lock(),mp.Value(c_int8,0)) for shm in shm_objects] #building args for our processes
    producer = mp.Process(target=producer_func,args=(buffers,))
    consumer = mp.Process(target=consumer_func,))
    consumer.start()
    producer.start()
    while True:
        try:
            time.sleep(1)
        except KeyboardInterrupt:
            break
    #signal child processes to close somehow
    #cleanup
    producer.join()
    consumer.join()
    for shm in shm_objects:
        shm.close()
        shm.unlink()