import threadingimport requestsfrom queue import Queue
class ImageDownloader:    def __init__(self, urls):        self.urls = urls        self.queue = Queue()
    def download_image(self, url):        response = requests.get(url)        if response.status_code == 200:            filename = url.split("/")[-1]            with open(filename, "wb") as f:                f.write(response.content)            print(f"Downloaded: {filename}")
    def worker(self):        while True:            url = self.queue.get()            if url is None:                break            self.download_image(url)            self.queue.task_done()
    def start_threads(self, num_threads=5):        threads = []        for _ in range(num_threads):            thread = threading.Thread(target=self.worker)            thread.start()            threads.append(thread)
        for url in self.urls:            self.queue.put(url)
        self.queue.join()
        for _ in range(num_threads):            self.queue.put(None)
        for thread in threads:            thread.join()
if __name__ == "__main__":    image_urls = ["url1", "url2", "url3", ...]  # 替换为实际图片的URL    downloader = ImageDownloader(image_urls)    downloader.start_threads()

import threading
counter = 0counter_lock = threading.Lock()
def increment_counter():    global counter    for _ in range(1000000):        with counter_lock:            counter += 1
def main():    thread1 = threading.Thread(target=increment_counter)    thread2 = threading.Thread(target=increment_counter)
    thread1.start()    thread2.start()
    thread1.join()    thread2.join()
    print("Counter:", counter)
if __name__ == "__main__":    main()

import threading
def thread_function():    try:        # 一些可能引发异常的操作        result = 10 / 0    except ZeroDivisionError as e:        print(f"Exception in thread: {e}")
if __name__ == "__main__":    thread = threading.Thread(target=thread_function)    thread.start()    thread.join()
    print("Main thread continues...")

import threadingimport time
class WorkerThread(threading.Thread):    def __init__(self, name, delay):        super().__init__()        self.name = name        self.delay = delay
    def run(self):        print(f"{self.name} started.")        time.sleep(self.delay)        print(f"{self.name} completed.")
if __name__ == "__main__":    thread1 = WorkerThread("Thread 1", 2)    thread2 = WorkerThread("Thread 2", 1)
    thread1.start()    thread2.start()
    thread1.join()    thread2.join()
    print("Main thread continues...")

import threadingimport time
class ResourceManager:    def __init__(self, total_resources):        self.total_resources = total_resources        self.available_resources = total_resources        self.lock = threading.Lock()
    def allocate(self, request):        with self.lock:            if self.available_resources >= request:                print(f"Allocated {request} resources.")                self.available_resources -= request            else:                print("Insufficient resources.")
    def release(self, release):        with self.lock:            self.available_resources += release            print(f"Released {release} resources.")
class UserThread(threading.Thread):    def __init__(self, name, resource_manager, request, release):        super().__init__()        self.name = name        self.resource_manager = resource_manager        self.request = request        self.release = release
    def run(self):        print(f"{self.name} started.")        self.resource_manager.allocate(self.request)        time.sleep(1)  # Simulate some work with allocated resources        self.resource_manager.release(self.release)        print(f"{self.name} completed.")
if __name__ == "__main__":    manager = ResourceManager(total_resources=5)
    user1 = UserThread("User 1", manager, request=3, release=2)    user2 = UserThread("User 2", manager, request=2, release=1)
    user1.start()    user2.start()
    user1.join()    user2.join()
    print("Main thread continues...")

import pdb
# 在代码中插入断点pdb.set_trace()

import timeit
def my_function():    # 要测试的代码
# 测试函数执行时间execution_time = timeit.timeit(my_function, number=1)print(f"Execution time: {execution_time} seconds")

import cProfile
def my_function():    # 要测试的代码
# 运行性能分析cProfile.run("my_function()")

# 安装line_profilerpip install line_profiler
# 使用line_profiler进行性能分析kernprof -l script.pypython -m line_profiler script.py.lprof

import asyncio
async def my_coroutine():    print("Start coroutine")    await asyncio.sleep(1)    print("Coroutine completed")
async def main():    await asyncio.gather(my_coroutine(), my_coroutine())
if __name__ == "__main__":    asyncio.run(main())

import asyncioimport aiohttp
async def download_image(session, url):    async with session.get(url) as response:        if response.status == 200:            filename = url.split("/")[-1]            with open(filename, "wb") as f:                f.write(await response.read())            print(f"Downloaded: {filename}")
async def main():    image_urls = ["url1", "url2", "url3", ...]  # 替换为实际图片的URL    async with aiohttp.ClientSession() as session:        tasks = [download_image(session, url) for url in image_urls]        await asyncio.gather(*tasks)
if __name__ == "__main__":    asyncio.run(main())

import asyncio
async def my_coroutine():    try:        # 异步操作        await asyncio.sleep(1)        raise ValueError("An error occurred")    except ValueError as e:        print(f"Caught an exception: {e}")
async def main():    task = asyncio.ensure_future(my_coroutine())    await asyncio.gather(task)
if __name__ == "__main__":    asyncio.run(main())