为何Python多实例比线程发送Scapy二层数据包速度更快？

我懂你现在的痛点——之前靠开多个命令行窗口跑同一个脚本实例，每个能稳定输出30Mbps左右的流量，三个实例加起来能冲到90Mbps，但现在想把这些逻辑整合到单个脚本里用线程实现，省得手动开多个窗口对吧？不过直接用多线程跑Scapy发包，很容易遇到资源竞争或者GIL限制的问题，导致单脚本的流量上不去，甚至不如多实例的效果。下面给你分析原因和几个可行的优化方案：

为什么多线程不如多实例？ #

• 多实例之所以能跑满带宽，是因为每个实例都有独立的Python解释器和独立的二层套接字资源，互相之间没有锁竞争，能充分利用网络接口的发包能力。
• 而单脚本多线程的话，如果多个线程共用同一个Scapy套接字，会因为套接字的内置锁导致线程互相等待；另外Python的GIL虽然对IO密集型任务影响不大，但Scapy的上层封装本身会有一些同步逻辑，也会拖慢并发效率。

优化方案1：每个线程创建独立的二层套接字 #

核心思路就是不让多个线程抢同一个套接字，每个线程初始化自己的L2Socket，这样每个线程都有独立的发包通道。示例代码如下：

from scapy.all import Ether, IP, UDP, conf, L2Socket
import threading
import time

def send_traffic(thread_id, packet, count, iface):
    # 每个线程单独创建二层套接字
    sock = L2Socket(iface=iface)
    print(f"线程{thread_id}启动，开始发包")
    start_time = time.time()
    
    for _ in range(count):
        sock.send(packet)
    
    end_time = time.time()
    # 计算速率
    total_bytes = len(packet) * count
    mbps = (total_bytes * 8) / (1024 * 1024 * (end_time - start_time))
    print(f"线程{thread_id}完成，速率：{mbps:.2f} Mbps")
    sock.close()

if __name__ == "__main__":
    # 替换成你的测试数据包和网卡名称
    target_iface = "eth0"  # Windows下可能是"以太网"这类名称
    test_packet = Ether()/IP(dst="192.168.1.100")/UDP(dport=5000)/("X"*100)
    
    # 创建3个线程，模拟之前多实例的效果
    threads = []
    for i in range(3):
        t = threading.Thread(
            target=send_traffic,
            args=(i, test_packet, 150000, target_iface)
        )
        threads.append(t)
        t.start()
    
    # 等待所有线程完成
    for t in threads:
        t.join()

优化方案2：用异步IO替代线程（更高效） #

如果你的Python版本在3.7以上，推荐用Scapy的异步发包能力，异步IO是单线程内的非阻塞操作，避免了线程切换的开销，效率比线程更高。示例代码：

from scapy.all import Ether, IP, UDP, L2Socket
import asyncio
import time

async def async_send_packets(sock, packet, count):
    print("异步任务启动")
    start_time = time.time()
    
    for _ in range(count):
        await sock.send(packet)
    
    end_time = time.time()
    total_bytes = len(packet) * count
    mbps = (total_bytes * 8) / (1024 * 1024 * (end_time - start_time))
    print(f"异步任务完成，速率：{mbps:.2f} Mbps")

async def main():
    target_iface = "eth0"
    test_packet = Ether()/IP(dst="192.168.1.100")/UDP(dport=5000)/("X"*100)
    
    # 创建一个异步二层套接字
    sock = L2Socket(iface=target_iface)
    # 启动3个异步发包任务
    tasks = [async_send_packets(sock, test_packet, 150000) for _ in range(3)]
    await asyncio.gather(*tasks)
    
    sock.close()

if __name__ == "__main__":
    asyncio.run(main())

优化方案3：绕过Scapy，用原生Raw Socket发包 #

如果Scapy的上层封装还是限制了速率，可以直接用Python的原生socket构造二层数据包发送，减少Scapy的额外开销，能获得更接近硬件上限的发包速率。示例：

import socket
import threading
import time
from scapy.all import Ether, IP, UDP

def raw_socket_send(thread_id, iface, packet_bytes, count):
    # 创建AF_PACKET类型的原始套接字，用于二层发包
    s = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(3))
    s.bind((iface, 0))
    
    print(f"线程{thread_id}启动，开始发包")
    start_time = time.time()
    
    for _ in range(count):
        s.send(packet_bytes)
    
    end_time = time.time()
    total_bytes = len(packet_bytes) * count
    mbps = (total_bytes * 8) / (1024 * 1024 * (end_time - start_time))
    print(f"线程{thread_id}完成，速率：{mbps:.2f} Mbps")
    s.close()

if __name__ == "__main__":
    target_iface = "eth0"
    # 用Scapy构造数据包，转成字节流
    test_packet = Ether()/IP(dst="192.168.1.100")/UDP(dport=5000)/("X"*100)
    packet_bytes = bytes(test_packet)
    
    threads = []
    for i in range(3):
        t = threading.Thread(
            target=raw_socket_send,
            args=(i, target_iface, packet_bytes, 150000)
        )
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()

额外注意事项 #

• 系统权限：不管用哪种方式，都需要以管理员/root权限运行脚本，因为原始套接字和二层发包需要特殊权限。
• 系统参数调整：Linux下可以调整套接字缓冲区大小来提升性能，执行以下命令：

  sysctl -w net.core.wmem_max=16777216
  sysctl -w net.core.rmem_max=16777216
  

• 网卡性能：确保你的网卡支持高流量发包，有些低端网卡可能会存在硬件瓶颈，限制最大发包速率。

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