结合Python多线程,和队列queue实现一个简单的队列消费者实例。用例批量检测ip是否存在。
from __future__ import print_function
import subprocess
import threading
from queue import Queue
from queue import Empty
def call_ping(ip):
try:
# subprocess.check_call检查命令是否执行正确
subprocess.check_call(['ping',ip], stdout=subprocess.PIPE)
print("{} is alive".format(ip))
except subprocess.CalledProcessError as e:
print("{} is unreachable.".format(ip))
print(e.output)
# print(e.returncode)
def is_reachable(q):
try:
while True:
# q.get_nowait非阻塞从队列取出元素
ip = q.get_nowait()
call_ping(ip)
except Empty:
pass
def main():
# 使用queue定义一个队列缓冲区,避免频繁的上下文切换
q = Queue()
with open('ip.txt', 'r', encoding='utf-8') as f:
for line in f.readlines():
# put添加元素
q.put(line.strip('\n'))
threads = []
for i in range(2):
thr = threading.Thread(target=is_reachable, args=[q,])
thr.start()
threads.append(thr)
for thr in threads:
thr.join()
if __name__ == '__main__':
main()