Python多线程数据处理实战技巧

import threading
import pandas as pd
from queue import Queue
<p>def load_csv_file(filepath, result_queue):
try:
df = pd.read_csv(filepath)
result_queue.put(df)
except Exception as e:
print(f"读取失败: {filepath}, 错误: {e}")</p><p>file_paths = ["data1.csv", "data2.csv", "data3.csv"]
result_queue = Queue()
threads = []</p><p>for path in file_paths:
t = threading.Thread(target=load_csv_file, args=(path, result_queue))
t.start()
threads.append(t)</p><p>for t in threads:
t.join()</p><h1>合并所有结果</h1><p>cleaned_data = pd.concat([result<em>queue.get() for </em> in range(result_queue.qsize())], ignore_index=True)</p>

from concurrent.futures import ThreadPoolExecutor
import requests
<p>def fetch_user_data(user_id):
url = f"<a target='_blank'  href='https://www.17golang.com/gourl/?redirect=MDAwMDAwMDAwML57hpSHp6VpkrqbYLx2eayza4KafaOkbLS3zqSBrJvPsa5_0Ia6sWuR4Juaq6t9nq5roGCUgXuytMyero2KedWwoYeYkbqVsJqthaW7ZGmosWyOqoqRg7Kx3LtrhaCN376Me8qZurWjgd99mqtkgqCzfZ-nfoCObbOqt7GBhpvOva56mYenz22Sun1ju6x9rr-Njpp-poZt' rel='nofollow'>https://api.example.com/users/{user_id</a>}"
response = requests.get(url)
if response.status_code == 200:
return response.json()
return None</p><p>user_ids = list(range(1, 101))  # 假设有100个用户需拉取</p><p>with ThreadPoolExecutor(max_workers=10) as executor:
results = list(executor.map(fetch_user_data, user_ids))</p><h1>过滤空值并转换为DataFrame</h1><p>valid_data = [r for r in results if r]
user_df = pd.DataFrame(valid_data)</p>