如何控制Python 3.11多进程消息发送速度，1分钟发送500k消息？

针对你遇到的多进程发送速率过高导致服务器宕机的问题，结合36核服务器的配置，这里提供几个可落地的方案：

1. 全局速率限流（精准控制总时长） #

核心思路是计算目标速率（500k条/60秒≈8333条/秒），通过多进程共享的计数器来控制全局发送节奏，确保总时长稳定在1分钟左右。

实现示例： #

import time
from multiprocessing import Process, Value, Lock

TARGET_TOTAL = 500000
TARGET_DURATION = 60  # 目标时长（秒）
TARGET_RATE = TARGET_TOTAL / TARGET_DURATION  # 约8333条/秒

def worker(shared_counter, lock, send_func):
    while True:
        with lock:
            if shared_counter.value >= TARGET_TOTAL:
                break
            current_count = shared_counter.value
            shared_counter.value += 1
        # 校准发送时间，控制全局速率
        expected_time = current_count / TARGET_RATE
        current_time = time.time()
        if current_time < expected_time:
            time.sleep(expected_time - current_time)
        # 执行你的发送逻辑
        send_func()

def main():
    shared_counter = Value('i', 0)
    lock = Lock()
    # 进程数建议控制在20-30之间（留资源给服务器系统进程）
    process_num = 24
    processes = []
    for _ in range(process_num):
        p = Process(target=worker, args=(shared_counter, lock, your_send_function))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()

if __name__ == "__main__":
    main()

2. 单进程限流+多进程分摊 #

如果不想用跨进程同步，可给每个进程分配固定任务量，让单进程独立控制发送节奏，总时长自然收敛到目标值。

实现示例： #

import time
from multiprocessing import Process

def worker(task_count, duration):
    interval = duration / task_count  # 单条消息的发送间隔
    for _ in range(task_count):
        your_send_function()
        time.sleep(interval)

def main():
    total_tasks = 500000
    target_duration = 60
    process_num = 20
    tasks_per_process = total_tasks // process_num
    processes = []
    # 启动主进程组
    for _ in range(process_num):
        p = Process(target=worker, args=(tasks_per_process, target_duration))
        processes.append(p)
        p.start()
    # 处理无法整除的剩余任务
    remaining = total_tasks % process_num
    if remaining > 0:
        p = Process(target=worker, args=(remaining, target_duration))
        p.start()
        processes.append(p)
    for p in processes:
        p.join()

if __name__ == "__main__":
    main()

3. 基于服务器负载动态调整进程数 #

如果想适配服务器实时状态，可结合psutil库监控CPU、网络负载，动态增减进程数，避免资源耗尽。

核心逻辑示例： #

import psutil
import time
from multiprocessing import Process, Manager

def worker(task_queue):
    while not task_queue.empty():
        task = task_queue.get()
        your_send_function(task)
        # 单进程内部加微小间隔，避免突发流量
        time.sleep(0.001)

def monitor_load(processes, task_queue, max_processes=30):
    while not task_queue.empty() or len(processes) > 0:
        cpu_usage = psutil.cpu_percent(interval=2)
        # CPU负载低则加进程，负载高则减进程
        if cpu_usage < 50 and len(processes) < max_processes:
            p = Process(target=worker, args=(task_queue,))
            p.start()
            processes.append(p)
        elif cpu_usage > 70 and len(processes) > 4:
            p = processes.pop()
            p.terminate()
            p.join()
        time.sleep(5)

def main():
    total_tasks = 500000
    manager = Manager()
    task_queue = manager.Queue()
    for _ in range(total_tasks):
        task_queue.put(your_task_data)
    # 初始启动8个进程
    processes = []
    for _ in range(8):
        p = Process(target=worker, args=(task_queue,))
        p.start()
        processes.append(p)
    # 启动负载监控进程
    monitor_proc = Process(target=monitor_load, args=(processes, task_queue))
    monitor_proc.start()
    monitor_proc.join()

if __name__ == "__main__":
    main()

额外建议 #

• 优先测试全局速率限流方案，它的时长控制最精准，能直接满足1分钟的要求
• 不要开满36核进程，留部分资源给服务器系统进程，建议进程数控制在20-30之间
• 如果是网络请求发送，尽量复用连接（比如用requests.Session），减少TCP握手开销，避免服务器端口耗尽

内容的提问来源于stack exchange，提问作者trung