timmy / 使用 ThreadPoolExecutor 進行多執行緒任務管理
0 喜欢
0 派生
1 文件
最后活跃于 9 months ago
ThreadPoolExecutor 提供了一種簡單的方式來管理多執行緒，適用於 I/O 密集型任務（如網路請求、檔案處理、資料庫查詢），提高執行效率。
| 1 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 2 | import time |
| 3 | |
| 4 | def task(n): |
| 5 | """模擬一個耗時任務""" |
| 6 | time.sleep(n) |
| 7 | return f"Task {n} completed after {n} seconds" |
| 8 | |
| 9 | # 建立執行緒池 |
| 10 | with ThreadPoolExecutor(max_workers=3) as executor: |
timmy / 使用 threading.Timer 延遲開啟網頁
0 喜欢
0 派生
1 文件
最后活跃于 9 months ago
此程式透過 threading.Timer 延遲指定秒數後，自動開啟指定的 URL。適用於自動化測試、提醒功能或延遲啟動特定網頁，確保主程式不中斷的情況下執行非同步操作。
| 1 | import threading |
| 2 | import webbrowser |
| 3 | |
| 4 | def open_url(url): |
| 5 | """ 開啟指定的 URL """ |
| 6 | webbrowser.open(url, new=2) |
| 7 | |
| 8 | # 設定延遲時間（秒） |
| 9 | delay = 2 |
| 10 | url = "https://www.example.com" |
timmy / 使用 asyncio 與 threading 進行非同步延遲執行
0 喜欢
0 派生
1 文件
最后活跃于 10 months ago
這段程式碼結合了 Python 的 asyncio 和 threading，透過 loop.run_in_executor 在背景執行同步的 blocking_sleep()，避免阻塞 asyncio 事件迴圈。這允許 async_sleep() 在等待時仍能執行其他非同步任務，適用於需要在 asyncio 應用程式中處理阻塞性 I/O 操作（如 time.sleep()）的情境。
| 1 | import asyncio |
| 2 | import threading |
| 3 | import time |
| 4 | |
| 5 | async def delayed_execution(): |
| 6 | # 在這裡執行延遲後的任務 |
| 7 | print("Delayed execution") |
| 8 | |
| 9 | def blocking_sleep(seconds): |
| 10 | time.sleep(seconds) |
timmy / 使用 Threading 進行非同步延遲執行
0 喜欢
0 派生
1 文件
最后活跃于 10 months ago
這段程式碼使用 Python 的 threading 模組來實現非同步延遲執行。當 async_sleep(seconds) 被呼叫時，它會建立一個新的執行緒來執行 sleep() 函式，而不會阻塞主執行緒。這允許主程式繼續執行其他任務，同時在指定時間後執行 delayed_execution()。這對於需要非同步延遲執行的應用場景，如計時器或背景任務，特別有用。
| 1 | import threading |
| 2 | import time |
| 3 | |
| 4 | def delayed_execution(): |
| 5 | # 在這裡執行延遲後的任務 |
| 6 | print("Delayed execution") |
| 7 | |
| 8 | def async_sleep(seconds): |
| 9 | def sleep(): |
| 10 | time.sleep(seconds) |
timmy / 多執行緒工作佇列處理
0 喜欢
0 派生
1 文件
最后活跃于 10 months ago
這段 Python 程式碼 使用 多執行緒 (Threading) 與工作佇列 (Queue) 來 處理 30 個工作項目，確保所有工作都執行完畢後，才會顯示 "All work completed" 訊息。
| 1 | """ |
| 2 | 這段程式碼是建立一個執行緒來處理工作項目，並將 30 個工作項目放入工作佇列中。然後會等待所有工作都完成後才會印出 "All work completed" 的訊息。 |
| 3 | """ |
| 4 | |
| 5 | import threading |
| 6 | import queue |
| 7 | |
| 8 | |
| 9 | # 定義一個函式來處理工作項目 |
| 10 | def process_work_item(work_queue): |
上一页
下一页