Python - 12. 동시성과 병렬 처리

Python에서는 동시성(Concurrency)과 병렬성(Parallelism)을 구현하기 위한 다양한 도구와 라이브러리를 제공합니다. 이 섹션에서는 멀티스레딩, 멀티프로세싱, 비동기 프로그래밍의 기초 및 고급 활용법을 살펴봅니다.

---

12.1 멀티스레딩

멀티스레딩이란?

멀티스레딩은 하나의 프로세스 내에서 여러 스레드가 작업을 수행하도록 합니다. 주로 I/O 바운드 작업에 적합합니다.

스레드 동기화

여러 스레드가 동일한 자원을 공유할 때, 동기화 기법을 사용하여 데이터 충돌을 방지해야 합니다.

import threading
import time

balance = 0
lock = threading.Lock()

def deposit(amount):
    global balance
    with lock:  # Lock을 사용하여 동기화
        local_balance = balance
        time.sleep(0.1)
        local_balance += amount
        balance = local_balance

threads = []
for _ in range(5):
    t = threading.Thread(target=deposit, args=(100,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Final balance: {balance}")

스레드 풀 사용

ThreadPoolExecutor를 사용하여 스레드를 효율적으로 관리할 수 있습니다.

from concurrent.futures import ThreadPoolExecutor

def task(n):
    print(f"Task {n} 시작")
    time.sleep(1)
    print(f"Task {n} 완료")

with ThreadPoolExecutor(max_workers=3) as executor:
    for i in range(5):
        executor.submit(task, i)

---

12.2 멀티프로세싱

멀티프로세싱이란?

멀티프로세싱은 여러 프로세스를 활용하여 병렬 처리를 수행합니다. 주로 CPU 바운드 작업에 적합합니다.

프로세스 간 데이터 공유

Value와 Array를 사용하여 데이터를 공유할 수 있습니다.

from multiprocessing import Process, Value, Array

shared_value = Value('i', 0)
shared_array = Array('i', [0, 0, 0])

def modify_data():
    shared_value.value += 1
    for i in range(len(shared_array)):
        shared_array[i] += 1

processes = [Process(target=modify_data) for _ in range(3)]
for p in processes:
    p.start()
for p in processes:
    p.join()

print(shared_value.value)  # 3
print(shared_array[:])  # [3, 3, 3]

Pool 사용

Pool은 작업을 병렬로 실행할 때 유용합니다.

from multiprocessing import Pool

def square(n):
    return n * n

with Pool(4) as pool:
    results = pool.map(square, [1, 2, 3, 4])

print(results)  # [1, 4, 9, 16]

---

12.3 비동기 프로그래밍 기초

비동기 프로그래밍이란?

비동기 프로그래밍은 작업이 완료될 때까지 기다리지 않고, 다른 작업을 수행할 수 있도록 합니다. 주로 I/O 바운드 작업에 적합합니다.

에러 처리 예제

import asyncio

async def task_with_error():
    try:
        raise ValueError("Error in task")
    except ValueError as e:
        print(f"Handled error: {e}")

asyncio.run(task_with_error())

타임아웃 처리

import asyncio

async def long_running_task():
    await asyncio.sleep(10)
    return "Task Completed"

async def main():
    try:
        result = await asyncio.wait_for(long_running_task(), timeout=5)
        print(result)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

동시성 제한

Semaphore를 사용하여 동시에 실행되는 작업의 개수를 제한할 수 있습니다.

import asyncio

semaphore = asyncio.Semaphore(2)

async def limited_task(name):
    async with semaphore:
        print(f"Task {name} 시작")
        await asyncio.sleep(2)
        print(f"Task {name} 완료")

async def main():
    tasks = [limited_task(i) for i in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())

---

12.4 asyncio와 async/await

asyncio 주요 기능

• 코루틴 실행
• 태스크(Task) 생성 및 관리
• 비동기 I/O 처리

태스크 예제

import asyncio

async def task1():
    await asyncio.sleep(2)
    print("Task 1 완료")

async def task2():
    await asyncio.sleep(1)
    print("Task 2 완료")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

---

12.5 비동기 웹 크롤링

aiohttp를 활용한 웹 크롤링

import asyncio
import aiohttp

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = ["https://example.com" for _ in range(5)]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        responses = await asyncio.gather(*tasks)
        for response in responses:
            print(response[:100])  # 응답 내용 일부 출력

asyncio.run(main())

---

12.6 비동기 데이터베이스 처리

asyncpg를 활용한 비동기 데이터베이스 처리

import asyncio
import asyncpg

async def fetch_data():
    conn = await asyncpg.connect(user="user", password="password", database="testdb", host="127.0.0.1")
    rows = await conn.fetch("SELECT * FROM my_table")
    for row in rows:
        print(row)
    await conn.close()

asyncio.run(fetch_data())

---

이번 섹션에서는 Python에서 동시성과 병렬 처리를 구현하는 다양한 방법과 활용 사례를 살펴보았습니다. 적절한 도구를 선택하여 프로그램의 성능과 효율성을 극대화할 수 있습니다.