如何在RxPY的map操作符中正确使用subprocess？

代码逻辑有两个核心错误，和异步机制、subprocess本身无关：

• Observable创建方法使用错误
  rx.of(ip_list)会把传入的整个IP列表作为单个值发射到数据流中，map操作符里接收到的ip参数根本不是单个IP字符串，而是完整的Python列表。
  • 当你使用简化版ping直接return ip时，subscribe回调里的list(x)是对整个列表做迭代，所以看起来能输出IP列表，实际上完全没有触发逐个IP的ping逻辑。
  • 当你使用真实ping逻辑时，subprocess.call的命令参数最后一位传入的是Python列表而非字符串，执行时会直接抛出类型错误，你没有捕获异常，程序在这行直接终止，自然走不到后面的print("Here!")，看起来就像这行代码被跳过了。
• 流处理逻辑不符合预期
  你需要实现的是逐个发射IP、逐个执行ping检测、最后收集存活IP的逻辑，而不是把整个列表作为单个值丢进流里只处理一次。

首先把Observable创建方法从rx.of换成rx.from_，这个方法会把可迭代对象里的元素逐个发射；其次调整流处理步骤，过滤掉ping失败返回的空字符串，最后把所有存活IP收集成列表再输出。

from reactivex import operators as ops
import reactivex as rx
import subprocess

def ping(ip):
    # 注意：Windows系统请替换ping参数为 ["ping", "-n", "1", "-w", "1000", ip]
    # 以下参数适配Linux/macOS系统
    retval = subprocess.call(
        ["ping", "-c1", "-n", "-i0.1", "-W1", ip],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )
    return ip if retval == 0 else ""

if __name__ == "__main__":
    ip_list = ["192.168.1.1", "192.168.1.2"]
    rx.from_(ip_list).pipe(
        ops.map(lambda ip: ping(ip)),
        ops.filter(lambda res: res != ""),
        ops.to_list()
    ).subscribe(lambda alive_ips: print("存活主机列表：", alive_ips))

上面的写法是同步逐个执行ping，扫描大段IP时速度很慢，可以搭配RxPY的线程池调度器实现并发检测，大幅提升扫描效率：

from reactivex import operators as ops
import reactivex as rx
from reactivex.scheduler import ThreadPoolScheduler
import subprocess
import multiprocessing

# 线程数和CPU核心数一致
thread_pool = ThreadPoolScheduler(multiprocessing.cpu_count())

def ping(ip):
    retval = subprocess.call(
        ["ping", "-c1", "-n", "-i0.1", "-W1", ip],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL
    )
    return ip if retval == 0 else ""

if __name__ == "__main__":
    # 示例扫描192.168.1.1~192.168.1.254网段
    ip_list = [f"192.168.1.{i}" for i in range(1, 255)]
    rx.from_(ip_list).pipe(
        ops.flat_map(lambda ip: rx.just(ip).pipe(
            ops.map(lambda x: ping(x)),
            ops.subscribe_on(thread_pool)
        )),
        ops.filter(lambda res: res != ""),
        ops.to_list()
    ).subscribe(lambda alive_ips: print("存活主机列表：", sorted(alive_ips)))

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