Python Language
멀티 스레딩

소개

멀티 스레딩의 기초

threading 모듈을 사용하여 새 threading.Thread 를 만들고 실행 함수를 할당하여 새 threading.Thread 시작할 수 있습니다.

import threading

def foo():
  print "Hello threading!"

my_thread = threading.Thread(target=foo)

target 매개 변수는 실행할 함수 또는 호출 가능 객체를 참조합니다. 스레드는 Thread 객체에서 start 가 호출 될 때까지 실행을 start 하지 않습니다.

스레드 시작

my_thread.start() # prints 'Hello threading!'

이제 my_thread 가 실행되고 종료 my_thread start 다시 호출하면 RuntimeError 가 생성됩니다. 스레드를 데몬으로 실행하려면 daemon=True kwarg를 전달하거나 start() 를 호출하기 전에 my_thread.daemon 을 True 설정하면 Thread 가 백그라운드에서 자동으로 데몬으로 실행됩니다.

주제에 참여하기

하나의 큰 작업을 여러 개의 작은 작업으로 나눠서 동시에 실행하고 싶지만 계속하기 전에 모든 작업이 끝날 때까지 기다려야하는 경우에는 Thread.join() 이 원하는 방법입니다.

예를 들어 웹 사이트의 여러 페이지를 다운로드하여 단일 페이지로 컴파일하려고한다고 가정 해 보겠습니다. 당신이 할거야 :

import requests
from threading import Thread
from queue import Queue

q = Queue(maxsize=20)
def put_page_to_q(page_num):
    q.put(requests.get('http://some-website.com/page_%s.html' % page_num)

def compile(q):
    # magic function that needs all pages before being able to be executed
    if not q.full():
        raise ValueError
    else:
        print("Done compiling!")

threads = []
for page_num in range(20):
     t = Thread(target=requests.get, args=(page_num,))
     t.start()
     threads.append(t)

# Next, join all threads to make sure all threads are done running before
# we continue. join() is a blocking call (unless specified otherwise using 
# the kwarg blocking=False when calling join)
for t in threads:
    t.join()

# Call compile() now, since all threads have completed
compile(q)

join() 어떻게 작동하는지 자세히 살펴 보시려면 여기를 클릭하십시오 .

사용자 지정 스레드 클래스 만들기

threading.Thread 클래스를 사용하여 새 사용자 정의 Thread 클래스를 하위 클래스화할 수 있습니다. 우리는 하위 클래스에서 run 메소드를 오버라이드해야한다.

from threading import Thread
import time

class Sleepy(Thread):

    def run(self):
        time.sleep(5)
        print("Hello form Thread")

if __name__ == "__main__":
    t = Sleepy()
    t.start()      # start method automatic call Thread class run method.
    # print 'The main program continues to run in foreground.'
    t.join()
    print("The main program continues to run in the foreground.")

스레드 간 통신

코드에는 여러 개의 스레드가 있으며 스레드간에 안전하게 통신해야합니다.

당신은 사용할 수있는 Queue 로부터 queue 라이브러리입니다.

from queue import Queue
from threading import Thread

# create a data producer 
def producer(output_queue):
    while True:
        data = data_computation()
        
        output_queue.put(data)

# create a consumer
def consumer(input_queue):
    while True:
        # retrieve data (blocking)
        data = input_queue.get()

        # do something with the data

        # indicate data has been consumed
        input_queue.task_done()

공유 대기열을 사용하여 생성자 및 소비자 스레드 만들기

q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()

작업자 풀 만들기

threading 및 queue :

from socket import socket, AF_INET, SOCK_STREAM
from threading import Thread
from queue import Queue
    
def echo_server(addr, nworkers):
    print('Echo server running at', addr)
    # Launch the client workers
    q = Queue()
    for n in range(nworkers):
        t = Thread(target=echo_client, args=(q,))
        t.daemon = True
        t.start()

    # Run the server
    sock = socket(AF_INET, SOCK_STREAM)
    sock.bind(addr)
    sock.listen(5)
    while True:
        client_sock, client_addr = sock.accept()
        q.put((client_sock, client_addr))

echo_server(('',15000), 128)

concurrent.futures.Threadpoolexecutor 사용 :

from socket import AF_INET, SOCK_STREAM, socket
from concurrent.futures import ThreadPoolExecutor

def echo_server(addr):
    print('Echo server running at', addr)
    pool = ThreadPoolExecutor(128)
    sock = socket(AF_INET, SOCK_STREAM)
    sock.bind(addr)
    sock.listen(5)
    while True:
        client_sock, client_addr = sock.accept()
        pool.submit(echo_client, client_sock, client_addr)

echo_server(('',15000))

David Beazley와 Brian K. Jones (O'Reilly)의 제 3 판, Python Cookbook. Copyright 2013 David Beazley and Brian Jones, 978-1-449-34037-7.

멀티 스레드 고급 사용

이 섹션에는 멀티 스레딩을 사용하여 구현 된 가장 고급 예제가 포함됩니다.

고급 프린터 (로거)

모든 것을 인쇄하는 스레드가 수신되어 터미널 너비에 따라 출력이 수정됩니다. 좋은 점은 터미널의 너비가 변경되면 "이미 작성된"출력이 수정된다는 것입니다.

#!/usr/bin/env python2

import threading
import Queue
import time
import sys
import subprocess
from backports.shutil_get_terminal_size import get_terminal_size

printq = Queue.Queue()
interrupt = False
lines = []

def main():

    ptt = threading.Thread(target=printer) # Turn the printer on
    ptt.daemon = True
    ptt.start()

    # Stupid example of stuff to print
    for i in xrange(1,100):
        printq.put(' '.join([str(x) for x in range(1,i)]))           # The actual way to send stuff to the printer
        time.sleep(.5)

def split_line(line, cols):
    if len(line) > cols:
        new_line = ''
        ww = line.split()
        i = 0
        while len(new_line) <= (cols - len(ww[i]) - 1):
            new_line += ww[i] + ' '
            i += 1
            print len(new_line)
        if new_line == '':
            return (line, '')

        return (new_line, ' '.join(ww[i:]))
    else:
        return (line, '')


def printer():

    while True:
        cols, rows = get_terminal_size() # Get the terminal dimensions
        msg = '#' + '-' * (cols - 2) + '#\n' # Create the
        try:
            new_line = str(printq.get_nowait())
            if new_line != '!@#EXIT#@!': # A nice way to turn the printer
                                         # thread out gracefully
                lines.append(new_line)
                printq.task_done()
            else:
                printq.task_done()
                sys.exit()
        except Queue.Empty:
            pass

        # Build the new message to show and split too long lines
        for line in lines:
            res = line          # The following is to split lines which are
                                # longer than cols.
            while len(res) !=0:
                toprint, res = split_line(res, cols)
                msg += '\n' + toprint

        # Clear the shell and print the new output
        subprocess.check_call('clear') # Keep the shell clean
        sys.stdout.write(msg)
        sys.stdout.flush()
        time.sleep(.5)

while 루프가있는 정지 가능한 스레드

import threading
import time

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self):
        super(StoppableThread, self).__init__()
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def join(self, *args, **kwargs):
        self.stop()
        super(StoppableThread,self).join(*args, **kwargs)

    def run()
        while not self._stop_event.is_set():
            print("Still running!")
            time.sleep(2)
        print("stopped!"

이 질문 에 기초.