파이썬 스레드 쓰레드 (thread) 및 대기열 (queue)

파이썬 스레드 쓰레드 (thread) 및 대기열 (queue)

---

기본적으로 파이썬에서 프로그램을 실행하면 싱글 쓰레드로 진행이 되기 때문에 코드의 순서에 따라서 움직이게 됩니다. 즉 특정 코드 구간에서 멈춰버리면 다음 코드 구간으로 넘어가서 작업을 하지 않는다는 것이죠.

 

따라서 병렬 처리를 하기 위해서는 별도의 모듈을 사용하여 구현해야 합니다. 병렬 처리(스레드)란 코드를 개별적으로 실행해서 대기한다는 개념 없이 한 번에 움직이게 하는 용도라고 볼 수 있습니다.

 

먼저 쓰레드 사용 예시 코드입니다.

import threading
from datetime import datetime
import time


def print_num():
    now_time = datetime.now().strftime('%H:%M:%S')
    print(now_time)
    time.sleep(1)


for go in range(10):
    threading.Thread(target=print_num).start()


==결과==
14:42:29
14:42:29
14:42:29
14:42:29
14:42:29
.
.
.

 

print_num 함수를 실행하면 현재 시간을 출력하게 됩니다. 그리고 아래 for문에서 쓰레드 모듈을 사용하여 print_num 함수를 10번 실행하였습니다.

 

원래 싱글 쓰레드의 경우라면 print_num 함수의 sleep에 의해서 1초간 대기 후 함수 리턴이 되기 때문에 결과 값은 1초씩 증가해야 하지만 위 코드에서는 병렬로 처리가 되었기 때문에 한 번에 모든 함수가 실행이 된 거죠.

 

쓰레드를 통한 함수에서 매개변수 값을 전달하기 위해서는 아래와 같은 방법을 쓸 수 있습니다.

def print_num(txt):
    now_time = datetime.now().strftime('%H:%M:%S')
    print(now_time + f' / {txt}')
    time.sleep(1)


for go in range(10):
    threading.Thread(target=print_num, kwargs={'txt': go + 1}).start()


==결과==
15:24:32 / 1
15:24:32 / 2
15:24:32 / 3
15:24:32 / 4
15:24:32 / 5

kwargs에 딕셔너리 형태로 함수의 매개변수에 값을 대입할 수 있습니다.

 

그리고 쓰레드는 별도의 설정을 하지 않는다면 메인 쓰레드가 종료되더라도 서브 스레드는 계속해서 구동이 됩니다. 만약 메인 쓰레드가 끝났을 때 진행 중이던 서브 쓰레드도 종료를 시키려면 아래와 같은 코드를 추가해주면 됩니다.

    th = threading.Thread(target=print_num)
    th.daemon = True
    th.start()

daemon을 True로 설정하면 해당 쓰레드는 메인 쓰레드가 종료될 때 같이 종료가 됩니다. daemon의 기본 설정은 False입니다.

 

이처럼 쓰레드를 잘 활용하면 작업이 되는 소요시간을 많이 줄일 수 있다는 장점이 있습니다. 그러나 이러한 쓰레드를 사용한 함수들이 데이터를 처리하는 과정에서 다른 쓰레드가 끝날 때 까지 기다려야 하는 상황이 생길 수가 있는데요. 이런 경우에는 쓰레드 내에서 대기열 기능을 추가하여 활용할 수 있습니다.

import threading
from datetime import datetime
import time
from queue import Queue


def print_num():
    q_text = q.get()  # q에 추가된 첫번째 요소 가져오기
    now_time = datetime.now().strftime('%H:%M:%S')
    print(f'{now_time} / {q_text}')
    time.sleep(1)
    q.task_done()  # q에서 가져온 항목 처리


q = Queue(maxsize=0)  # 대기열 객체 생성

for go in range(10):
    q.put(f'q에 데이터 추가!: {go + 1}')
    threading.Thread(target=print_num).start()
    q.join()  # q의 모든 항목이 처리되면 루프 탈출됨


==결과==
15:03:23 / q에 데이터 추가!: 1
15:03:24 / q에 데이터 추가!: 2
15:03:25 / q에 데이터 추가!: 3
15:03:26 / q에 데이터 추가!: 4
15:03:27 / q에 데이터 추가!: 5

기존 코드에서 queue라는 모듈만 추가가 되었습니다. 대기열 모듈인데 예시에서는 q라는 변수에 객채를 생성하여 코드를 작성하였습니다. 대기열을 사용하면 쓰레드로 함수를 실행하더라도 기존 쓰레드의 리턴 여부를 체크하고 처리가 완료될 때까지 대기를 하는 것입니다. 그리고 다음 쓰레드로 넘어가는 것이죠.

 

이러한 대기열은 어떻게 사용을 하는 것인지 간단하게 설명을 해보도록 하겠습니다.

from queue import Queue


q = Queue(maxsize=0)  # 대기열 객체 생성

for go in range(5):
    # q 객체에 추가할 데이터
    q.put(f'데이터: {go}')

for p in q.queue:
    print(p)

==결과==
데이터: 0
데이터: 1
데이터: 2
데이터: 3
데이터: 4

대기열은 put이라는 함수를 사용하여 특정 데이터를 대기열 객체에 넣을 수 있습니다. 그리고 데이터는 리스트 형식으로 대기열 객체에 저장이 됩니다. 위 결과를 보면 for 반복문에서 put이 사용되어 총 5개의 데이터가 q 객체에 저장이 된 것을 볼 수 있습니다.

q = Queue(maxsize=0)  # 대기열 객체 생성

for go in range(5):
    # q 객체에 추가할 데이터
    q.put(f'데이터: {go}')

a = q.get()  # 첫번째 데이터를 pop 방식으로 가져온다
print(a + ' / 빠져나온 첫번째 데이터\n')

for p in q.queue:
    print(p)

==결과==
데이터: 0 / 빠져나온 첫번째 데이터

데이터: 1
데이터: 2
데이터: 3
데이터: 4

대기열에서 get은 put으로 저장된 데이터를 pop 형식으로 가져오는 것입니다. q라는 객체에서 첫 번째 데이터를 제외시키면서 반환이 됩니다.

from queue import Queue


q = Queue(maxsize=0)  # 대기열 객체 생성
q.put('test1')

a = q.get()
print(a)

q.task_done()  # q에서 가져온 항목 처리

q.join()  # q의 모든 항목이 처리되면 루프 탈출됨
print('완료!')

==결과==
test1
완료!

put으로 q객체에 데이터를 넣고 get으로 첫 번째 객체를 불러오고 다음으로 사용되는 것은 task_done입니다. 이 함수는 q객체에서 특정 데이터를 사용 완료할 때마다 의무적으로 넣어주며 q객체를 탈출하기 위한 용도입니다.

 

join 함수는 q에 있는 모든 데이터를 사용하였을 때 탈출을 시키는 용도입니다. q 객체에 있던 모든 데이터가 처리되었을 때 다음 코드가 진행되는 것이죠.

 

정리를 하자면 put에 데이터를 저장하고 get으로 데이터 반환, task_done으로 데이터 사용 마무리가 됩니다. 그리고 단지 대기열을 쓰레드 함수의 리턴 체크 용도로만 사용한다고 하면 get까지 사용할 필요는 없습니다. put에 임의 데이터를 넣어주고 task_done만 처리해줘도 join에서 대기열 탈출이 가능합니다.

 

그렇기 때문에 만약 put으로 저장된 데이터가 2개인데 task_done을 한 번만 사용했다고 하면 join에 도달이 되더라도 다음 코드는 진행되지 않습니다.

q = Queue(maxsize=0)  # 대기열 객체 생성
q.put('test1')
q.put('test2')

a = q.get()
print(a)

q.task_done()  # q에서 가져온 항목 처리

q.join()  # q의 모든 항목이 처리되면 루프 탈출됨
print('완료!')

==결과==
test1

이처럼 put으로 q라는 객체에는 두 개의 데이터가 들어왔는데 task_done으로 항목 처리가 된 것은 한 개뿐이니 아직 데이터를 다 활용하지 않은 상태라는 것에서 완료에 도달을 할 수 없는 것입니다.

스레드와 대기열을 적절하게 잘 활용한다면 보다 수준이 높은 코드를 작성할 수 있으며 코드의 진행 속도에도 큰 차이를 보여주기 때문에 데이터 처리가 오래 걸리는 작업에서는 시스템 부하가 걸리지 않는 선에서 적당히 스레드를 활용하는 것이 좋습니다.