개요
오늘은 GUI 애플리케이션을 개발할 때 자주 겪는 스레드 종료 문제 에 대해 알아보려고 한다.
보통 QThread를 활용한 백그라운드 작업을 모두 수행하고나서
애플리케이션 종료할 때, 스레드가 제대로 종료되지 않아 프로그램이 "응답 없음" 상태가 되는 경우가 발생한다.
이번 포스팅에서는 그 문제를 해결하는 방법을 기록할 것이다.
문제 상황
GUI 애플리케이션을 개발할 때, 백그라운드에서 지속적으로 작업을 수행하는 스레드를 구현하는 경우가 많다.
(웹 사이트에서 주기적으로 데이터를 가져오기, 등등 ..)
그러나 애플리케이션을 종료할 때, 실행 중인 스레드가 즉시 종료되지 않고 오랜 시간 동안 대기 상태에 머물러 "응답 없음" 상태가 되는 문제가 발생할 수 있다.
이는 주로 스레드 내에서 긴 time.sleep() 등의 호출로 인해 종료 요청을 즉시 반영하지 못하기 때문이다.
이러한 문제를 해결하기 위해서는 스레드가 종료 요청을 신속하게 감지하고 안전하게 종료될 수 있도록 설계해야 한다.
기존 코드 분석
문제("응답 없음")를 일으키는 프로그램을 간단하게 구현해보자.
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtCore import QThread, Signal
from threading import Event
import time
import requests
class ScrapingThread(QThread):
new_post_signal = Signal(str)
error_signal = Signal(str)
def __init__(self, login_info, comment_text):
super().__init__()
self.pause_event = Event()
self.pause_event.set()
self.running = True
self.login_info = login_info
self.comment_text = comment_text
self.session = requests.Session()
self.logged_in = False
def run(self):
try:
if not self.logged_in:
self.perform_login()
while self.running:
self.pause_event.wait() # 일시정지 상태에서는 대기
new_post = self.check_new_post()
if new_post:
self.add_comment(new_post)
time.sleep(60) # 1분 대기
except Exception as e:
self.error_signal.emit(str(e))
# 핵심 로직 구현 ...위 코드에서 run 메소드는 무한 루프를 돌며 새로운 게시글을 확인하고 댓글을 작성한 후, time.sleep(60)으로 1분 동안 대기한다.
그러나 애플리케이션을 종료할 때, 이 time.sleep(60) 동안 스레드는 종료 요청을 즉시 처리하지 못하고 최대 60초 동안 대기하게 되어 "응답 없음" 상태가 된다.
QThread의 인터럽트 매커니즘 사용하기
이 문제를 해결하기 위해 QThread의 인터럽트 매커니즘 을 활용할 수 있다.
QThread는 requestInterruption()과 isInterruptionRequested() 메소드를 제공하여 스레드에 종료 요청을 보낼 수 있다.
이를 통해 스레드가 종료 요청을 감지하고 적절히 종료할 수 있게된다.
주요 메소드 설명
requestInterruption(): 스레드에 인터럽트 요청을 보냄isInterruptionRequested(): 스레드가 인터럽트 요청을 받았는지 확인함
코드 수정하기
이제 문제가 발생하는 기존 코드를 수정하여 QThread의 인터럽트 매커니즘을 적용해보자.
from PySide6.QtCore import QThread, Signal
from threading import Event
import time
import requests
class ScrapingThread(QThread):
new_post_signal = Signal(str)
error_signal = Signal(str)
def __init__(self, login_info, comment_text):
super().__init__()
self.pause_event = Event()
self.pause_event.set()
self.login_info = login_info
self.comment_text = comment_text
self.session = requests.Session()
self.logged_in = False
def run(self):
try:
if not self.logged_in:
self.perform_login()
while not self.isInterruptionRequested():
self.pause_event.wait() # 일시정지 상태에서는 대기
if self.isInterruptionRequested():
break # 인터럽트 요청 시 루프 종료
new_post = self.check_new_post()
if new_post:
self.add_comment(new_post)
# 60초 동안 1초 간격으로 대기하여 종료 요청을 신속히 반영
for _ in range(60):
if self.isInterruptionRequested():
break
time.sleep(1)
except Exception as e:
self.error_signal.emit(str(e))
def perform_login(self):
# 로그인 로직 구현
login_url = "https://example.com/login"
payload = {
'username': self.login_info['username'],
'password': self.login_info['password']
}
response = self.session.post(login_url, data=payload, timeout=10)
if response.status_code == 200:
self.logged_in = True
self.new_post_signal.emit("로그인 성공")
else:
raise Exception("로그인 실패")
def check_new_post(self):
# 게시글 확인 로직 구현
check_url = "https://example.com/api/posts"
response = self.session.get(check_url, timeout=10)
if response.status_code == 200:
posts = response.json()
# 새로운 게시글인지 판단하는 로직 구현
# 여기서는 예시로 항상 새로운 게시글이 있다고 가정
return posts[0]['id']
else:
raise Exception("게시글 확인 실패")
def add_comment(self, post_id):
# 댓글 작성 로직 구현
comment_url = f"https://example.com/api/posts/{post_id}/comments"
payload = {
'comment': self.comment_text
}
response = self.session.post(comment_url, data=payload, timeout=10)
if response.status_code == 201:
self.new_post_signal.emit(f"게시글 {post_id}에 댓글 작성 완료")
else:
raise Exception("댓글 작성 실패")
def pause(self):
self.pause_event.clear()
def resume(self):
self.pause_event.set()주요 변경 사항
- 인터럽트 요청 확인:
while not self.isInterruptionRequested(): 조건을 사용하여 인터럽트 요청이 있는지 확인한다. - 짧은 간격으로 대기:
time.sleep(60)대신 1초 간격으로 60번 반복하며 인터럽트 요청을 감지할 수 있도록 변경하였다. 이를 통해 최대 60초의 지연 없이 종료 요청을 반영할 수 있게 된다. - 인터럽트 요청 시 루프 종료: 루프 내에서
if self.isInterruptionRequested(): break를 추가하여 인터럽트 요청 시 즉시 루프를 종료하도록 변경하였다.
UI 앱 코드는 다음과 같다.
from PySide6.QtWidgets import QApplication, QWidget, QPushButton, QVBoxLayout, QLabel, QLineEdit
from PySide6.QtCore import QThread, Signal
from threading import Event
import time
import requests
class CommentAutomationApp(QWidget):
def __init__(self):
super().__init__()
self.init_ui()
self.scraping_thread = None # 스레드 초기화
self.login_info = {'username': 'your_username', 'password': 'your_password'}
self.comment_text = "자동 댓글입니다."
def init_ui(self):
self.setWindowTitle("댓글 자동화 프로그램")
self.setGeometry(100, 100, 300, 250)
# 상태 표시 레이블
self.status_label = QLabel("상태: 로그인 필요", self)
# 댓글 내용 입력 필드
self.comment_input = QLineEdit(self)
self.comment_input.setPlaceholderText("댓글 내용을 입력하세요")
# 버튼 생성
self.login_button = QPushButton("로그인", self)
self.start_button = QPushButton("시작", self)
self.pause_button = QPushButton("일시정지", self)
self.resume_button = QPushButton("재개", self)
# 초기 버튼 상태 설정
self.start_button.setEnabled(False)
self.pause_button.setEnabled(False)
self.resume_button.setEnabled(False)
# 버튼 클릭 이벤트 연결
self.login_button.clicked.connect(self.login)
self.start_button.clicked.connect(self.start_scraping)
self.pause_button.clicked.connect(self.pause_scraping)
self.resume_button.clicked.connect(self.resume_scraping)
# 레이아웃 설정
layout = QVBoxLayout()
layout.addWidget(self.status_label)
layout.addWidget(self.comment_input)
layout.addWidget(self.login_button)
layout.addWidget(self.start_button)
layout.addWidget(self.pause_button)
layout.addWidget(self.resume_button)
self.setLayout(layout)
def login(self):
# 로그인 로직 구현
self.comment_text = self.comment_input.text() or "자동 댓글입니다."
self.scraping_thread = ScrapingThread(self.login_info, self.comment_text)
self.scraping_thread.new_post_signal.connect(self.update_status)
self.scraping_thread.error_signal.connect(self.handle_error)
try:
self.scraping_thread.perform_login()
self.status_label.setText("상태: 로그인 성공")
self.start_button.setEnabled(True)
except Exception as e:
self.status_label.setText(f"상태: 로그인 실패 - {e}")
def start_scraping(self):
if self.scraping_thread and not self.scraping_thread.isRunning():
self.scraping_thread.start()
self.status_label.setText("상태: 자동화 시작")
self.start_button.setEnabled(False)
self.pause_button.setEnabled(True)
self.login_button.setEnabled(False)
def pause_scraping(self):
if self.scraping_thread and self.scraping_thread.isRunning():
self.scraping_thread.pause()
self.status_label.setText("상태: 자동화 일시정지")
self.pause_button.setEnabled(False)
self.resume_button.setEnabled(True)
def resume_scraping(self):
if self.scraping_thread and self.scraping_thread.isRunning():
self.scraping_thread.resume()
self.status_label.setText("상태: 자동화 재개")
self.pause_button.setEnabled(True)
self.resume_button.setEnabled(False)
def update_status(self, message):
self.status_label.setText(f"상태: {message}")
def handle_error(self, error_message):
self.status_label.setText(f"오류: {error_message}")
self.start_button.setEnabled(False)
self.pause_button.setEnabled(False)
self.resume_button.setEnabled(False)
self.login_button.setEnabled(True)
def closeEvent(self, event):
# 프로그램 종료 시 스레드 안전하게 종료
if self.scraping_thread and self.scraping_thread.isRunning():
self.scraping_thread.requestInterruption() # 인터럽트 요청
self.scraping_thread.wait() # 스레드가 종료될 때까지 대기
event.accept()closeEvent메소드: 애플리케이션 종료 시requestInterruption()을 호출하여 스레드에 인터럽트 요청을 보낸다. 그 후wait()을 호출하여 스레드가 종료될 때까지 기다린다- 스레드 상태 관리: 스레드의 시작, 일시정지, 재개 버튼의 상태를 적절히 관리하여 사용자가 원하는 대로 스레드를 제어할 수 있도록 하였다.