Rumah > pembangunan bahagian belakang > Tutorial Python > Saya menggunakan kelas subproses.Popen dengan benang tetapi ia tidak berfungsi untuk saya

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

WBOY
Lepaskan: 2024-02-22 13:13:11
ke hadapan
716 orang telah melayarinya

我正在使用带有线程的 subprocess.Popen 类,但它对我不起作用

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
Salin selepas log masuk

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)
Salin selepas log masuk

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)
    Salin selepas log masuk

    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
    Salin selepas log masuk

    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
    Salin selepas log masuk
  • 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
Salin selepas log masuk

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!

sumber:stackoverflow.com
Kenyataan Laman Web ini
Kandungan artikel ini disumbangkan secara sukarela oleh netizen, dan hak cipta adalah milik pengarang asal. Laman web ini tidak memikul tanggungjawab undang-undang yang sepadan. Jika anda menemui sebarang kandungan yang disyaki plagiarisme atau pelanggaran, sila hubungi admin@php.cn
Tutorial Popular
Lagi>
Muat turun terkini
Lagi>
kesan web
Kod sumber laman web
Bahan laman web
Templat hujung hadapan