[C++] 22. 스레드 풀(Thread Pool), 생산자 소비자 패턴(Producer-consumer)

 

지난 시간에서는 스레드 사용할 때 주의할 점과 스레드의 실제 사용 예시를 살펴봤습니다. [C++] 21. 스레드 경쟁 상태(Race condition), 뮤텍스(mutex), 데드락, std::conditional_variable, std::async

이번 시간에는 스레드 패키지의 마지막 시간, 스레드 풀에 대해서 살펴보시죠!

 

 

 

생산자-소비자 패턴(Producer-Consumer)

Producer-Consumer 패턴은 소프트웨어 디자인에서 사용되는 중요한 디자인 패턴 중 하나로, 다른 컴포넌트 간의 작업을 분리하고 조절하는 데 사용됩니다. 이 패턴은 다수의 생산자(Producer)와 소비자(Consumer) 간의 협력을 효과적으로 관리하는 데 사용됩니다.

생산자 소비자 패턴은 데이터의 진입(income)과 처리(process)가 병행적으로 진행될 때 사용합니다. 예를 들어, 한 스레드에서 데이터를 생산(or 이벤트 수신)하고 다른 스레드에서 이 데이터를 처리하는 경우에 유용합니다.

생산자-소비자 패턴의 구성 요소와 역할

1. 생산자(Producer): 데이터를 생성하고 공유 데이터 구조(예: 큐 또는 버퍼)에 데이터를 넣습니다.
2. 소비자(Consumer): 공유 데이터 구조에서 데이터를 가져와 처리합니다.
3. 공유 데이터 구조(Shared Data Structure): 생산자와 소비자가 데이터를 주고받는 공통의 저장소입니다. 이 구조는 스레드 간 동기화를 관리하고 데이터를 안전하게 보관하는 역할을 합니다.
4. 동기화 메커니즘(Synchronization Mechanism): 스레드 간의 안전한 데이터 공유를 위해 사용됩니다. 대표적으로 뮤텍스(mutex)나 세마포어(semaphore)와 같은 동기화 도구를 활용합니다.

 

출처: https://coding-food-court.tistory.com/81

 

생산자-소비자 패턴의 장점

- 성능 향상: 데이터 생성과 처리를 병렬로 실행하여 시스템 성능을 향상시킵니다.
- 데이터 처리 분리: 데이터 생성과 처리 로직을 분리함으로써 코드 유지 관리와 확장성을 향상시킵니다.
- 자원 관리: 공유 자원에 대한 안전한 접근을 보장하여 경합 조건(race condition) 및 데드락(deadlock) 같은 다중 스레드 문제를 방지합니다.

Producer-Consumer 패턴은 다양한 언어와 환경에서 구현될 수 있으며, 멀티스레딩, 멀티프로세싱, 분산 시스템 등에서 사용됩니다. 이 패턴을 잘 이해하고 활용하면 복잡한 다중 스레드 환경에서도 안정적이고 효율적인 소프트웨어를 개발하는 데 도움이 됩니다.

 

 

 

 

스레드풀(Thread Pool)

스레드 풀(Thread Pool)은 다중 스레드 환경에서 작업을 관리하고 실행하는 데 사용되는 소프트웨어 스킬입니다. 주로 멀티스레드 서버 애플리케이션과 같이 동시에 여러 작업을 처리해야 하는 상황에서 유용하게 활용됩니다. 앞서 소개한 생산자-소비자 패턴을 스레드풀로 구현할 예정입니다.

스레드 풀의 구성 요소

1. 스레드 집합(workers): 스레드 풀은 미리 생성된 스레드 집합(workers)을 가지고 있습니다. 이 스레드들은 작업을 실행할 준비가 되어 있습니다. 스레드 풀의 크기는 일반적으로 설정 가능하며, 이 크기는 시스템 자원과 작업 부하에 따라 조절할 수 있습니다.

2. 작업 큐(job queue): 스레드 풀은 작업을 저장하는 큐를 관리합니다. 이 큐는 스레드 풀에 제출된 작업들을 저장하고 대기시킵니다. 작업은 큐에 추가되면 스레드 풀 내의 스레드들이 순서대로 작업을 가져와 실행합니다.

3. 작업 제출 인터페이스(inqueue): 클라이언트 애플리케이션은 스레드 풀에 작업을 제출하는 인터페이스를 사용하여 작업을 스레드 풀에 추가할 수 있습니다. 작업은 스레드 풀에 제출되면 큐에 추가되어 스레드 풀 내의 스레드들이 차례대로 처리합니다.

출처: https://tecoble.techcourse.co.kr/post/2021-09-18-java-thread-pool/

 

 

 

 

스레드 풀 구현

먼저 스레드 풀을 구현하기 위한 기술은 std::thread, std::condition_variable, std::mutex, std::queue, std::vector, template 정도가 있습니다. 초보자분들은 구현하기 쉽지 않지만 스레드 풀의 경우 한 번만 이해하고 작성한다면 어디서든 사용 가능한 무기가 될 것입니다.

#include <iostream>
#include <thread>
#include <vector>
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>

class ThreadPool {
public:
    ThreadPool(size_t numThreads) : stop(false) {
        for (size_t i = 0; i < numThreads; ++i) {
            workers.emplace_back([this] {
                while (true) {
                    std::function<void()> task;

                    {
                        std::unique_lock<std::mutex> lock(queueMutex);
                        condition.wait(lock, [this] { return stop || !tasks.empty(); });

                        if (stop && tasks.empty()) {
                            return;
                        }

                        task = std::move(tasks.front());
                        tasks.pop();
                    }

                    task();
                }
            });
        }
    }

    template <class F>
    void enqueue(F&& func) {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            tasks.emplace(std::forward<F>(func));
        }
        condition.notify_one();
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queueMutex);
            stop = true;
        }
        condition.notify_all();

        for (std::thread& worker : workers) {
            worker.join();
        }
    }

private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;

    std::mutex queueMutex;
    std::condition_variable condition;
    bool stop;
};

int main() {
    ThreadPool pool(10); // 10개의 스레드를 가진 스레드 풀 생성

    // 작업을 스레드 풀에 추가
    for (int i = 0; i < 30; ++i) {
        pool.enqueue([i] {
            std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << std::endl;
        });
    }

    // 모든 작업이 완료될 때까지 대기
    std::this_thread::sleep_for(std::chrono::seconds(2));

    return 0;
}

 

간단한 스레드풀을 구현한 예시입니다. 스레드풀의 생성자에는 생성할 스레드의 개수를 입력받아서 개수만큼 스레드를 만듭니다. 스레드가 실행되는 함수는 람다함수로 즉시 만들어서 호출했습니다. 생성된 스레드에서는 notify를 대기하고 있다가 notify가 되면 잡큐(Job-queue)에서 task를 빼서(front와 pop) 처리하는 방식입니다. 생산자 부분(enqueue 함수)에서는 큐에 실행될 내용을 넣고 기다리고 있는 workers(대기 스레드)에게 notify 통지를 날립니다.

 

스레드 풀의 소멸자에서는 대기하고 있는 모든 스레드들을 깨워서 join하기 위해 stop플래그를 켬과 동시에 notify_all을 실행합니다. 이후 모든 workers에 대해 join으로 스레드를 소멸시킵니다. 위 예제의 Task에서는 간단히 스레드 id와 함께 메세지를 출력했습니다. 10개의 스레드를 생성시켰고, 30개의 task를 실행시켰습니다.

 

 

만약 여러분이 대규모 서비스를 운영한다고 가정했을 때, 동시에 들어오는 요청이 많을 것입니다. 이에 대해서 빠르게 처리하기 위해서는 병렬처리는 필수입니다. 그리고 프로세스의 자원을 고려했을 때 스레드 풀이 어떤 것 보다도 효과적입니다. 대규모 요청이 들어오더라도, 생산자는 Task를 큐에 넣기만 하고, 소비자(workers)가 Task를 처리함으로써 이를 소화할 수 있습니다. 이처럼 스레드풀은 프로그램의 성능을 향상시키고 시스템 안정성을 유지하는 데 도움이 될 수 있습니다.

 

 

---

 

 

 

다음 포스팅에서는 템플릿(Template)에 대해 알아보겠습니다^^

도움이 되셨길 바랄께요!