Rumah >pembangunan bahagian belakang >Tutorial Python >Saya menggunakan kelas subproses.Popen dengan benang tetapi ia tidak berfungsi untuk saya
Saya mempunyai fail berikut:
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)
Matlamat program ini adalah untuk mengetahui sama ada peranti rangkaian boleh diping. Saya menjalankan fail main.py dan ia memberikan hasil yang salah. Saya cuba menyahpepijat dan ia nampaknya masalah benang, tetapi saya tidak dapat menyelesaikannya.
Sebarang bantuan akan sangat dihargai
Beberapa pemerhatian
Seperti yang dinyatakan sebelum ini, anda terus menambah senarai ping_cmd
追加更多内容,您的命令将拥有越来越多的 ip 地址。最好每次构建一个新命令,而不是附加到 ping_cmd
Pertimbangkan baris ini
is_successes = findall("ttl", out)
Saya melihat output linux dan ia kelihatan seperti ini:
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
Dalam kod anda, anda mencari ttl (huruf besar) yang akan sentiasa menghasilkan false
. Selain itu, untuk mencari teks, anda tidak perlu menggunakan ungkapan biasa:
is_success = "ttl" in out
Memandangkan anda tidak menyiarkan utilities
包的内容,因此我不知道 packet_cout
apakah yang setara
Apabila berurusan dengan multi-threading, lebih baik menggunakan logging
daripada mencetak kerana yang pertama selamat untuk benang manakala yang kedua tidak.
Dengan itu, inilah cadangan penyelesaian saya
# 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
Atas ialah kandungan terperinci Saya menggunakan kelas subproses.Popen dengan benang tetapi ia tidak berfungsi untuk saya. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!