带有请求的 Python 多线程

Question

我有一个使用 IPV6 启动“请求”会话并获取一些数据的刮板，我现在有 10000 个 ip 列表，我已经使用线程对其进行了准备，但它给出了错误。 需要支持来找出问题。

import requests, queue,threading, urllib3,jso,pandas as pd, os, time, datetime,inspect
num_threads = 2
root = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
with open (root+ "/ip_list.txt") as ips:
    device_ip = list(ips)
class Writer_Worker(threading.Thread):
    def __init__(self, queue, df, *args, **kwargs):
        if not queue:
            print("Device Queue not specified")
            exit(1)
        self.out_q = queue
        self.df = df
        super().__init__(*args, **kwargs)
    def run(self):
        while True:
            try:
                device_details = self.out_q.get(timeout=3)
            except queue.Empty:
                return
            self.df[device_details[0]] = device_details
            self.out_q.task_done()
class Worker(threading.Thread):
    def __init__(self, queue, out_queue, device_password, *args, **kwargs):
        if not queue:
            print("Device Queue not specified")
            exit(1)
        self.queue = queue
        self.pas = device_password
        self.out_q = out_queue
        super().__init__(*args, **kwargs)
    def run(self):
        while True:
            try:
                device_ip = self.queue.get(timeout=3)
            except queue.Empty:
                return
            self.connect_to_device_and_process(device_ip)
            self.queue.task_done()
    def connect_to_device_and_process(self, device_ip):
        st = str("Online")
        try:
            r = requests.post("https://["+device_ip+"]/?q=index.login&mimosa_ajax=1", {"username":"configure", "password":self.pas}, verify=False)
        except requests.exceptions.ConnectionError:
                st = str("Offline")
                self.out_q.put([device_ip,st,"","","","","","","","","","","","","","","","","",""])
                return
        finally:
            if 'Online' in st:
                r = requests.get("https://["+device_ip+"]/cgi/dashboard.php", cookies=r.cookies, verify=False)
                if "Response [401]" in str(r):
                    st2 = str("Password Error")
                    self.out_q.put([device_ip,st2,"","","","","","","","","","","","","","","","","",""])
                else:
                    data = json.loads(r.content.decode())
                    output5 = data ['config'] ['Spectrum_Power']
                    self.out_q.put([device_ip,st,output5['Auto_Power'].replace('2', 'Max Power').replace('1', 'Min Power').replace('0', 'off'),output5['AutoConfig']])
def main():
    start = time.time()
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    pas = input("Enter Device Password:")
    df =pd.DataFrame(columns = ["IP","Status","Auto_Power","AutoChannel"])
    q = queue.Queue(len(device_ip))
    for ip in device_ip:
        q.put_nowait(ip)
    out_q = queue.Queue(len(device_ip))
    Writer_Worker(out_q, df).start()
    for _ in range(num_threads):
        Worker(q, out_q, pas).start()
    q.join()
    print(df)
    df.to_excel('iBridge_C5x_Audit_Report.xlsx', sheet_name='Detail', index = False)
    
if __name__ == "__main__":
    main()

以下是运行脚本时出现的错误，我无法登录到此设备。 任何帮助都是可观的。

Answer 1

您应该使用在固定数量的线程之间分配工作的线程池。这是 Python 3.2 版本以来的核心特性。

1. from concurrent.futures import ThreadPoolExecutor
2. 定义一个函数perform(ip)，执行一个ip的请求
3. 将变量 numThreads 设置为所需的线程数
4. 运行线程池执行器：

print(f'Using {numThreads} threads')
with ThreadPoolExecutor(max_workers=numThreads) as pool:
    success = all(pool.map(perform, ips))

来源：https://docs.python.org/3/library/concurrent.futures.html

在该页面上，您可以找到更适合您的应用程序的示例：https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor-example

Answer 2

从线程导入线程

th = 线程(target=self.fill_imdb, args=(movies_info_part, "thread" + str(count))) th.start()

fill_imdb 是我的方法