Python利用多线程优化for循环的技巧分享

import threading
 
# 方法一：继承threading.Thread类
class MyThread(threading.Thread):
    def __init__(self, name):
        threading.Thread.__init__(self)
        self.name = name
 
    def run(self):
        print(f"Starting {self.name}")
        # 在这里执行线程的任务
        print(f"Exiting {self.name}")
 
# 方法二：使用threading.Thread的构造函数
def thread_function(name):
    print(f"Starting {name}")
    # 在这里执行线程的任务
    print(f"Exiting {name}")
 
thread1 = MyThread("Thread-1")
thread2 = threading.Thread(target=thread_function, args=("Thread-2",))
 
thread1.start()
thread2.start()
 
thread1.join()
thread2.join()

import threading
 
lock = threading.Lock()
 
def thread_safe_function(name):
    with lock:
        print(f"Thread {name} is accessing the resource.")
        # 在这里执行线程安全的操作
 
threads = []
for i in range(5):
    thread = threading.Thread(target=thread_safe_function, args=(i,))
    threads.append(thread)
    thread.start()
 
for thread in threads:
    thread.join()

import threading
import requests
 
urls = [
    "http://www.example.com",
    "http://www.nonexistent-domain.com",
    # ...更多URL
]
 
def check_url(url):
    try:
        response = requests.get(url, timeout=5)
        print(f"{url} is {response.status_code}")
    except requests.RequestException as e:
        print(f"{url} failed: {e}")
 
threads = []
for url in urls:
    thread = threading.Thread(target=check_url, args=(url,))
    threads.append(thread)
    thread.start()
 
for thread in threads:
    thread.join()

import concurrent.futures
import requests
 
urls = [
    "http://www.example.com",
    "http://www.nonexistent-domain.com",
    # ...更多URL
]
 
def check_url(url):
    try:
        response = requests.get(url, timeout=5)
        return f"{url} is {response.status_code}"
    except requests.RequestException as e:
        return f"{url} failed: {e}"
 
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    future_to_url = {executor.submit(check_url, url): url for url in urls}
    for future in concurrent.futures.as_completed(future_to_url):
        url = future_to_url[future]
        try:
            result = future.result()
            print(result)
        except Exception as exc:
            print(f"{url} generated an exception: {exc}")

import time
import threading
import concurrent.futures
import requests
 
urls = [
    # 这里添加大量URL
] * 100  # 假设我们有100个相同的URL列表，以模拟大量数据
 
def single_threaded_check():
    for url in urls:
        check_url(url)
 
def multi_threaded_check():
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        future_to_url = {executor.submit(check_url, url): url for sublist in urls for url in sublist}
        for future in concurrent.futures.as_completed(future_to_url):
            url = future_to_url[future]
            try:
                future.result()
            except Exception as exc:
                print(f"{url} generated an exception: {exc}")
 
def check_url(url):
    try:
        response = requests.get(url, timeout=1)  # 缩短超时时间以模拟I/O密集型任务
    except requests.RequestException:
        pass
 
start_time = time.time()
single_threaded_check()
end_time = time.time()
print(f"Single-threaded execution time: {end_time - start_time} seconds")
 
start_time = time.time()
multi_threaded_check()
end_time = time.time()
print(f"Multi-threaded execution time: {end_time - start_time} seconds")