掌握 Python 并发编程：利用先进技术提升性能

import asyncioimport aiohttpasync def fetch_url(session, url):    async with session.get(url) as response:        return await response.text()async def main():    urls = ['http://example.com', 'http://example.org', 'http://example.net']    async with aiohttp.clientsession() as session:        tasks = [fetch_url(session, url) for url in urls]        results = await asyncio.gather(*tasks)        for url, result in zip(urls, results):            print(f"content length of {url}: {len(result)}")asyncio.run(main())

import concurrent.futuresimport requestsdef download_file(url):    response = requests.get(url)    filename = url.split('/')[-1]    with open(filename, 'wb') as f:        f.write(response.content)    return f"downloaded {filename}"urls = [    'https://example.com/file1.pdf',    'https://example.com/file2.pdf',    'https://example.com/file3.pdf']with concurrent.futures.threadpoolexecutor(max_workers=3) as executor:    future_to_url = {executor.submit(download_file, url): url for url in urls}    for future in concurrent.futures.as_completed(future_to_url):        url = future_to_url[future]        try:            data = future.result()        except exception as exc:            print(f"{url} generated an exception: {exc}")        else:            print(data)

import concurrent.futuresimport mathdef is_prime(n):    if n < 2:        return false    for i in range(2, int(math.sqrt(n)) + 1):        if n % i == 0:            return false    return truedef find_primes(start, end):    return [n for n in range(start, end) if is_prime(n)]ranges = [(1, 25000), (25001, 50000), (50001, 75000), (75001, 100000)]with concurrent.futures.processpoolexecutor() as executor:    results = executor.map(lambda r: find_primes(*r), ranges)all_primes = [prime for sublist in results for prime in sublist]print(f"found {len(all_primes)} prime numbers")

from multiprocessing import process, arrayimport numpy as npdef worker(shared_array, start, end):    for i in range(start, end):        shared_array[i] = i * iif __name__ == '__main__':    size = 10000000    shared_array = array('d', size)    # create 4 processes    processes = []    chunk_size = size // 4    for i in range(4):        start = i * chunk_size        end = start + chunk_size if i < 3 else size        p = process(target=worker, args=(shared_array, start, end))        processes.append(p)        p.start()    # wait for all processes to finish    for p in processes:        p.join()    # convert shared array to numpy array for easy manipulation    np_array = np.frombuffer(shared_array.get_obj())    print(f"sum of squares: {np_array.sum()}")

from threading import lock, threadclass counter:    def __init__(self):        self.count = 0        self.lock = lock()    def increment(self):        with self.lock:            self.count += 1def worker(counter, num_increments):    for _ in range(num_increments):        counter.increment()counter = counter()threads = []for _ in range(10):    t = thread(target=worker, args=(counter, 100000))    threads.append(t)    t.start()for t in threads:    t.join()print(f"final count: {counter.count}")

import asyncioimport aiohttpfrom asyncio import semaphoreasync def fetch_url(url, semaphore):    async with semaphore:        async with aiohttp.clientsession() as session:            async with session.get(url) as response:                return await response.text()async def main():    urls = [f'http://example.com/{i}' for i in range(100)]    semaphore = semaphore(10)  # limit to 10 concurrent connections    tasks = [fetch_url(url, semaphore) for url in urls]    results = await asyncio.gather(*tasks)    print(f"fetched {len(results)} urls")asyncio.run(main())

import asyncioasync def risky_operation(i):    if i % 2 == 0:        raise valueerror(f"even number not allowed: {i}")    await asyncio.sleep(1)    return iasync def main():    tasks = [risky_operation(i) for i in range(10)]    results = await asyncio.gather(*tasks, return_exceptions=true)    for result in results:        if isinstance(result, exception):            print(f"got an exception: {result}")        else:            print(f"got a result: {result}")asyncio.run(main())

import asyncioasync def fetch_data(url):    print(f"fetching data from {url}")    await asyncio.sleep(2)  # simulate network delay    return f"data from {url}"async def process_data(data):    print(f"processing {data}")    await asyncio.sleep(1)  # simulate processing time    return f"processed {data}"async def save_result(result):    print(f"saving {result}")    await asyncio.sleep(0.5)  # simulate saving delay    return f"saved {result}"async def fetch_process_save(url):    data = await fetch_data(url)    processed = await process_data(data)    return await save_result(processed)async def main():    urls = ['http://example.com', 'http://example.org', 'http://example.net']    tasks = [fetch_process_save(url) for url in urls]    results = await asyncio.gather(*tasks)    for result in results:        print(result)asyncio.run(main())

import asyncioasync def long_running_task(n):    print(f"Starting long task {n}")    try:        await asyncio.sleep(10)        print(f"Task {n} completed")        return n    except asyncio.CancelledError:        print(f"Task {n} was cancelled")        raiseasync def main():    tasks = [long_running_task(i) for i in range(5)]    try:        results = await asyncio.wait_for(asyncio.gather(*tasks), timeout=5)    except asyncio.TimeoutError:        print("Operation timed out, cancelling remaining tasks")        for task in tasks:            task.cancel()        # Wait for all tasks to finish (they'll raise CancelledError)        await asyncio.gather(*tasks, return_exceptions=True)    else:        print(f"All tasks completed successfully: {results}")asyncio.run(main())