我正在使用带有线程的 subprocess.Popen 类,但它对我不起作用
我有以下文件:
ping.py:
from utilities.env import packet_cout, ping_time, max_workers_count from utilities.cmds import create_ping_cmd from subprocess import popen, pipe, devnull, timeoutexpired from re import findall from concurrent.futures import threadpoolexecutor max_ping_time = float(packet_cout) * (float(ping_time)) ping_cmd = create_ping_cmd() def check_ping_process(ping_process): ping_result = {} try: out = ping_process.communicate(timeout=max_ping_time)[0] is_successes = findall("ttl", out) if ping_process.returncode == 0 and is_successes: return true else: return false except timeoutexpired: ping_process.kill() return false def ping_ip(ip): ping_cmd.append(ip) ping_process = popen( ping_cmd, stdout=pipe, stderr=devnull, text=true ) result = check_ping_process(ping_process) return result def ping_ip_list(ip_list): ping_results = [] with threadpoolexecutor(max_workers=max_workers_count) as executor: results = executor.map(ping_ip, ip_list) for ip, result in zip(ip_list, results): print({ip, result}) ping_results.append({ip, result}) return ping_results
main.py:
from ping import ping_ip_list ip_list = [ '192.168.0.100', '192.168.0.1', '192.168.0.104', '192.168.0.124', '192.168.0.103' ] def monitor_network_devices(ip_list): results = ping_ip_list(ip_list) monitor_network_devices(ip_list)
该程序的目标是查明网络设备是否可 ping 通。我正在运行 main.py 文件,它给出了错误的结果。我尝试debug,似乎是线程的问题,但无法解决。
任何帮助将不胜感激
正确答案
一些观察
如前所述,您不断向
ping_cmd
追加更多内容,您的命令将拥有越来越多的 ip 地址。最好每次构建一个新命令,而不是附加到ping_cmd
列表考虑这一行
is_successes = findall("ttl", out)
登录后复制我看一下 linux 的输出,它看起来像这样:
ping 192.168.1.1 (192.168.1.1) 56(84) bytes of data. 64 bytes from 192.168.1.1: icmp_seq=1 ttl=63 time=11.9 ms 64 bytes from 192.168.1.1: icmp_seq=2 ttl=63 time=5.18 ms --- 192.168.1.1 ping statistics --- 2 packets transmitted, 2 received, 0% packet loss, time 1024ms rtt min/avg/max/mdev = 5.182/8.541/11.901/3.359 ms
登录后复制在您的代码中,您搜索 ttl(大写),这将始终导致
false
。此外,要搜索文本,您不需要使用正则表达式:is_success = "ttl" in out
登录后复制由于您没有发布
utilities
包的内容,因此我不知道packet_cout
等值是什么处理多线程时,最好使用
logging
而不是 print,因为前者是线程安全的,而后者则不然。
这样,这是我提出的解决方案
# ping.py import logging import subprocess from concurrent.futures import ThreadPoolExecutor logging.basicConfig( level=logging.DEBUG, format="%(asctime)s | %(levelname)s | %(threadName)-15s | %(funcName)-18s | %(message)s", ) def ping_ip(ip): ping_command = ["ping", "-c1", ip] logging.debug("Execute command %r", ping_command) completed_process = subprocess.run( ping_command, text=True, capture_output=True, check=False, ) success = completed_process.returncode == 0 logging.debug("%s -> %r", ip, success) return success def ping_ip_list(ip_list): with ThreadPoolExecutor() as executor: out = dict(zip(ip_list, executor.map(ping_ip, ip_list))) return out
以上是我正在使用带有线程的 subprocess.Popen 类,但它对我不起作用的详细内容。更多信息请关注PHP中文网其他相关文章!

热AI工具

Undresser.AI Undress
人工智能驱动的应用程序,用于创建逼真的裸体照片

AI Clothes Remover
用于从照片中去除衣服的在线人工智能工具。

Undress AI Tool
免费脱衣服图片

Clothoff.io
AI脱衣机

Video Face Swap
使用我们完全免费的人工智能换脸工具轻松在任何视频中换脸!

热门文章

热工具

记事本++7.3.1
好用且免费的代码编辑器

SublimeText3汉化版
中文版,非常好用

禅工作室 13.0.1
功能强大的PHP集成开发环境

Dreamweaver CS6
视觉化网页开发工具

SublimeText3 Mac版
神级代码编辑软件(SublimeText3)

Linux终端中查看Python版本时遇到权限问题的解决方法当你在Linux终端中尝试查看Python的版本时,输入python...

使用FiddlerEverywhere进行中间人读取时如何避免被检测到当你使用FiddlerEverywhere...

在使用Python的pandas库时,如何在两个结构不同的DataFrame之间进行整列复制是一个常见的问题。假设我们有两个Dat...

Uvicorn是如何持续监听HTTP请求的?Uvicorn是一个基于ASGI的轻量级Web服务器,其核心功能之一便是监听HTTP请求并进�...

如何在10小时内教计算机小白编程基础?如果你只有10个小时来教计算机小白一些编程知识,你会选择教些什么�...

攻克Investing.com的反爬虫策略许多人尝试爬取Investing.com(https://cn.investing.com/news/latest-news)的新闻数据时,常常�...
