Saya menggunakan kelas subproses.Popen dengan benang tetapi ia tidak berfungsi untuk saya

Kandungan soalan

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

Jawapan betul

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!