### Python 并发处理 Buffer 限制任务数 为了优化 Python 代码并实现并发处理，同时限制同时运行的任务数量为 32 个，可以考虑使用 `asyncio` 或者 `concurrent.futures.ThreadPoolExecutor` 来管理任务队列。以下是两种方法的具体实现： #### 方法一：基于 `asyncio` 和 `Semaphore` `asyncio.Semaphore` 可用于控制并发执行的任务数量。通过设置信号量的最大值来限制同时运行的任务数目。 ```python import asyncio class HtmlLoader: def __init__(self, max_concurrency=32): self.semaphore = asyncio.Semaphore(max_concurrency) async def fetch_html(self, url): async with self.semaphore: # 控制最大并发数 print(f"Fetching {url}") await asyncio.sleep(1) # 模拟网络请求延迟 return f"HTML content from {url}" async def load_urls(self, urls): tasks = [self.fetch_html(url) for url in urls] results = await asyncio.gather(*tasks) return results if __name__ == "__main__": urls = [f"http://example.com/{i}" for i in range(300)] loader = HtmlLoader() loop = asyncio.get_event_loop() html_contents = loop.run_until_complete(loader.load_urls(urls)) ``` 此代码片段展示了如何利用异步编程模型和信号量来限制并发度[^1]。 #### 方法二：基于 `ThreadPoolExecutor` 和 `Queue` 如果更倾向于多线程方式，则可结合 `concurrent.futures.ThreadPoolExecutor` 和 `queue.Queue` 实现类似的逻辑。 ```python from concurrent.futures import ThreadPoolExecutor, as_completed import queue def fetch_html(url): print(f"Fetching {url}") import time time.sleep(1) # 模拟网络请求延迟 return f"HTML content from {url}" def load_urls_with_threadpool(urls, max_workers=32): result_queue = queue.Queue() def task_wrapper(url): result = fetch_html(url) result_queue.put(result) with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(task_wrapper, url) for url in urls] for future in as_completed(futures): try: future.result() except Exception as exc: print(f'Generated an exception: {exc}') while not result_queue.empty(): yield result_queue.get() if __name__ == "__main__": urls = [f"http://example.com/{i}" for i in range(300)] html_contents_generator = load_urls_with_threadpool(urls) for html_content in html_contents_generator: pass # 处理 HTML 内容 ``` 这种方法适用于 I/O 密集型操作，并且能够有效避免因过多线程引发的资源争用问题[^1]。 对于内存不足的情况，建议监控程序运行期间的实际内存消耗情况，并适当调整每次加载的数据量或者增加垃圾回收频率以释放不再使用的对象空间[^3]。