Python Language
multithreading

Invoering

Basisprincipes van multithreading

Met behulp van de threading module kan een nieuwe thread van uitvoering worden gestart door een nieuwe threading.Thread maken. Thread en deze een functie toewijzen om uit te voeren:

import threading

def foo():
  print "Hello threading!"

my_thread = threading.Thread(target=foo)

Het target parameter verwijst naar de functie (of opvraagbaar object) worden uitgevoerd. De thread begint pas met de uitvoering als start wordt aangeroepen op het Thread object.

Een draad beginnen

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

Nu my_thread is uitgevoerd en beëindigd, zal het opnieuw start een RuntimeError produceren. Als je je thread als een daemon wilt uitvoeren, als je de daemon=True kwarg doorgeeft of my_thread.daemon op True my_thread.daemon voordat je start() my_thread.daemon , wordt je Thread stil op de achtergrond uitgevoerd als een daemon.

Deelnemen aan een discussie

In gevallen waarin u een grote taak opsplitst in meerdere kleine en ze tegelijkertijd wilt uitvoeren, maar moet wachten tot ze allemaal zijn Thread.join() voordat u doorgaat, is Thread.join() de methode die u zoekt.

Stel bijvoorbeeld dat u meerdere pagina's van een website wilt downloaden en deze wilt samenvoegen tot één pagina. Je zou dit doen:

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)

Meer informatie over hoe join() werkt, vindt u hier .

Maak een aangepaste threadklasse

Met behulp van threading.Thread class kunnen we een nieuwe aangepaste Thread-klasse indelen. we moeten de run methode in een subklasse overschrijven.

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.")

Communiceren tussen threads

Uw code bevat meerdere threads en u moet veilig met elkaar communiceren.

U kunt een Queue uit de 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()

Producent- en consumententhreads maken met een gedeelde wachtrij

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

Een werknemerspool maken

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)

Gebruik van 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))

Python Cookbook, 3e editie, door David Beazley en Brian K. Jones (O'Reilly). Copyright 2013 David Beazley en Brian Jones, 978-1-449-34037-7.

Geavanceerd gebruik van multithreads

Deze sectie bevat enkele van de meest geavanceerde voorbeelden die zijn gerealiseerd met Multithreading.

Geavanceerde printer (logger)

Er wordt een thread ontvangen die alles afdrukt en de uitvoer aanpast aan de terminalbreedte. Het leuke is dat ook de "reeds geschreven" uitvoer wordt gewijzigd wanneer de breedte van de terminal verandert.

#!/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)

Stopbare draad met een tijdje lus

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!"

Gebaseerd op deze vraag .