Ich verwende die Klasse subprocess.Popen mit Threads, aber sie funktioniert bei mir nicht

Frageninhalt

Ich habe folgende Dateien:

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)

Das Ziel dieses Programms besteht darin, herauszufinden, ob ein Netzwerkgerät pingbar ist. Ich führe die Datei main.py aus und sie liefert falsche Ergebnisse. Ich habe versucht zu debuggen und es scheint ein Thread-Problem zu sein, aber ich kann es nicht lösen.

Jede Hilfe wäre sehr dankbar

Richtige Antwort

Einige Beobachtungen

• Wie bereits erwähnt, fügt man der ping_cmd 追加更多内容，您的命令将拥有越来越多的 ip 地址。最好每次构建一个新命令，而不是附加到 ping_cmdListe

  immer wieder etwas hinzu

• Betrachten Sie diese Zeile

  is_successes = findall("ttl", out)

  Ich schaue mir die Ausgabe von Linux an und sie sieht so aus:

  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

  In Ihrem Code suchen Sie nach ttl (Großbuchstaben), was immer zu false führt. Außerdem müssen Sie zum Durchsuchen von Text keine regulären Ausdrücke verwenden:

  is_success = "ttl" in out

• Da du nichts gepostet hast utilities 包的内容，因此我不知道 packet_cout was ist das Äquivalent

• Beim Umgang mit Multithreading ist es besser, logging anstelle von print zu verwenden, da ersteres threadsicher ist, letzteres jedoch nicht.

Damit ist hier mein Lösungsvorschlag

# 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

Das obige ist der detaillierte Inhalt vonIch verwende die Klasse subprocess.Popen mit Threads, aber sie funktioniert bei mir nicht. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!