import threading

def print_numbers():
    for i in range(1, 6):
        print(i)

# 創(chuàng)建線程
thread = threading.Thread(target=print_numbers)

# 啟動線程
thread.start()

# 等待線程完成
thread.join()

print("線程執(zhí)行完畢")

import threading

counter = 0
lock = threading.Lock()

def increment_counter():
    global counter
    with lock:
        counter += 1

threads = []
for _ in range(100):
    thread = threading.Thread(target=increment_counter)
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"計數(shù)器最終值: {counter}")

import threading
import queue

def producer(q):
    for i in range(5):
        q.put(i)
        print(f"生產(chǎn): {i}")

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消費: {item}")

q = queue.Queue()
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))

producer_thread.start()
consumer_thread.start()

producer_thread.join()
q.put(None)  # 發(fā)送結束信號
consumer_thread.join()

from concurrent.futures import ThreadPoolExecutor

def square(n):
    return n * n

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(square, i) for i in range(10)]
    results = [future.result() for future in futures]

print(results)

from concurrent.futures import ThreadPoolExecutor, as_completed

def factorial(n):
    if n == 0:
        return 1
    else:
        return n * factorial(n-1)

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(factorial, i) for i in range(10)]
    for future in as_completed(futures):
        print(f"結果: {future.result()}")
 

from concurrent.futures import ThreadPoolExecutor

def risky_task(n):
    if n == 5:
        raise ValueError("模擬異常")
    return n * 2

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(risky_task, i) for i in range(10)]
    for future in futures:
        try:
            result = future.result()
            print(f"結果: {result}")
        except Exception as e:
            print(f"任務執(zhí)行失敗: {e}")

import threading
import requests

def download(url):
    response = requests.get(url)
    print(f"下載 {url} 的內(nèi)容長度: {len(response.content)}")

urls = [
    "http://example.com",
    "http://example.org",
    "http://example.net"
]

threads = []
for url in urls:
    thread = threading.Thread(target=download, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

from concurrent.futures import ProcessPoolExecutor

def factorial(n):
    if n == 0:
        return 1
    else:
        return n * factorial(n-1)

with ProcessPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(factorial, range(20)))

print(results)

import threading
import requests

def download(url):
    response = requests.get(url)
    print(f"下載 {url} 的內(nèi)容長度: {len(response.content)}")

urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.github.com"
]

threads = []
for url in urls:
    thread = threading.Thread(target=download, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有下載任務完成")

from concurrent.futures import ThreadPoolExecutor

def fetch_url(url):
    response = requests.get(url)
    return len(response.content)

urls = [
    "https://www.example.com",
    "https://www.python.org",
    "https://www.github.com"
]

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(fetch_url, url): url for url in urls}
    for future in concurrent.futures.as_completed(futures):
        url = futures[future]
        try:
            data_length = future.result()
            print(f"{url} 的內(nèi)容長度: {data_length}")
        except Exception as exc:
            print(f"{url} 下載時發(fā)生錯誤: {exc}")

print("所有任務完成")

import threading
import queue

def worker(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"處理項目: {item}")
        q.task_done()

task_queue = queue.Queue()
num_worker_threads = 4

threads = []
for _ in range(num_worker_threads):
    thread = threading.Thread(target=worker, args=(task_queue,))
    thread.start()
    threads.append(thread)

for item in range(20):
    task_queue.put(item)

# 等待所有任務完成
task_queue.join()

# 停止工作線程
for _ in range(num_worker_threads):
    task_queue.put(None)
for thread in threads:
    thread.join()

print("所有任務處理完成")

import threading
import sqlite3

def query_database(db_name, query):
    conn = sqlite3.connect(db_name)
    cursor = conn.cursor()
    cursor.execute(query)
    result = cursor.fetchall()
    print(f"查詢結果: {result}")
    conn.close()

db_name = 'example.db'
queries = [
    "SELECT * FROM users",
    "SELECT * FROM orders",
    "SELECT * FROM products"
]

threads = []
for query in queries:
    thread = threading.Thread(target=query_database, args=(db_name, query))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有數(shù)據(jù)庫查詢完成")

import threading
from PIL import Image, ImageFilter

def process_image(image_path):
    img = Image.open(image_path)
    img = img.filter(ImageFilter.BLUR)
    output_path = f"blurred_{image_path}"
    img.save(output_path)
    print(f"{image_path} 已處理并保存為 {output_path}")

image_paths = [
    "image1.jpg",
    "image2.jpg",
    "image3.jpg"
]

threads = []
for image_path in image_paths:
    thread = threading.Thread(target=process_image, args=(image_path,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print("所有圖像處理完成")