J'utilise la classe subprocess.Popen avec des threads mais cela ne fonctionne pas pour moi

Contenu de la question

J'ai les fichiers suivants :

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)

Le but de ce programme est de savoir si un périphérique réseau peut être pingé. J'exécute le fichier main.py et il donne des résultats erronés. J'ai essayé le débogage et cela semble être un problème de thread, mais je n'arrive pas à le résoudre.

Toute aide serait grandement appréciée

Réponse correcte

Quelques observations

• Comme mentionné précédemment, vous continuez à ajouter à la ping_cmd 追加更多内容，您的命令将拥有越来越多的 ip 地址。最好每次构建一个新命令，而不是附加到 ping_cmd liste

• Considérez cette ligne

  is_successes = findall("ttl", out)

  Je regarde le résultat de Linux et cela ressemble à ceci :

  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

  Dans votre code, vous recherchez ttl (majuscule) qui donnera toujours false. De plus, pour rechercher du texte, vous n'avez pas besoin d'utiliser des expressions régulières :

  is_success = "ttl" in out

• Puisque tu n'as pas posté utilities 包的内容，因此我不知道 packet_cout quel est l'équivalent

• Lorsqu'il s'agit de multi-threading, il est préférable d'utiliser logging au lieu de print car le premier est thread-safe tandis que le second ne l'est pas.

Sur ce, voici ma solution proposée

# 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

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!