Thread간 정보 공유를 위한 Queue

Queue는 복수의 정보를 저장할 수 있는 저장공간이다. 리스트와 유사한 개념이므로 리스트로 Queue를 구현해 사용할 수도 있다. queue는 단일 쓰레드에서 사용할 수도 있는데, 그보다는 멀티쓰레드 그 진가를 발휘한다.  패키지 queue에서 제공하는 Queue 클래스는 thread간 정보를 공유할 수 있도록 쓰레드락을 사용한다. 우리는 이것을 이용하여 멀리쓰레드 환경에서 큐를 안전하게 사용할 수 있다. 

다음과 같은 작업을 가정하자.

메인쓰레드에서는 작업할 데이터를 추가하기만 한다. worker라는 서브쓰레드를 두어 메인쓰레드에서 추가한 데이터를 처리하는 일을 도맏는다. 두 작업은 디커플되어 있고, 서로에게 작업지연현상을 야기시켜서는 안된다. 

다음은, 리스트로 구현된 큐를 사용한 코드이다.

import threading
import time


def worker():
    while True:
        time.sleep(0.01)
        try:
            item = q.pop(0)  # FIFO queue. 자료가 없으면 IndexError: pop from empty list
            print(f'{item} 을 처리합니다.')
            if item == 'q':
                break
        except Exception:
            continue


q = ['아름다운', '우리나라', '대한민국']
threading.Thread(target=worker,).start()

while True:
    data = input('q or text :')
    q.append(data)
    print(f'q={q}')
    if data == 'q':
        break

 

다음은 `queue` 를 활용한 코드이다.

import threading, queue, time

def worker(q:queue.Queue, quit:threading.Event):
    while True:
        item = q.get()  # block = Ture, timeout=None  --> 큐에 항목이 있어서 반환될 때 까지 기다림(블럭킹)
        # block = Ture, timeout=10  --> 큐에 항목이 있어서 반환될 때 까지 기다림(블럭킹) 10초되면 Empty예외발생
        # block = Flase  --> 큐에 항목이 있어서 반환되지 않으면 Empty 예외 발생
        #time.sleep(0.01)   # 이 쓰레드는 1초동안 CPU사용하지 않음
        print(f'{item} 을 처리합니다. ')
        if quit.is_set() : break

q = queue.SimpleQueue()
for data in ['아름다운', '우리나라', '대한민국']:
    q.put(data)
quitSignal = threading.Event()
threading.Thread(target=worker, args=(q, quitSignal)).start()

while True:
    data = input('q or text :')
    q.put(data)
    if data == 'q':
        quitSignal.set()
        break

위 코드에서는 쓰레드간 신호를 전달하기 위해 threading.Event도 사용되었다. 무한루프가 있는 쓰레드를 정지시키기 위한 quitSignal 이벤트객체를 만들고 이를 통해 쓰레드정지 명령을 인가한다. 이벤트는 쓰레드 간에 디커플링을 보장한다. (pyqt5의 signal - slot 의 signal은 시그널을 주고 받는 객체간 커플링되어있어 시그널을 발생한 쓰레드에서는 해당슬롯이 종료될 때 까지 블럭된다.)

반면에 threading.Event는 set()함수를 호출하여 시그널을 발생시켜도 시그널을 받는 쓰레드의 작동여부와 상관없이 코드는 계속 진행된다.