Creating Custom Thread Classes

Python's threading module provides the ability to create custom thread classes by extending the base Thread class. This approach offers better encapsulation of data and behavior.

from threading import Thread
from time import sleep

def process_item(item_id):
    print(f'Processing item {item_id}')
    sleep(2)
    print(f'Item {item_id} completed')

class WorkerThread(Thread):
    def __init__(self, item_id):
        super(WorkerThread, self).__init__()
        self.item_id = item_id

    def run(self):
        process_item(self.item_id)

if __name__ == '__main__':
    thread = WorkerThread(1)
    thread.start()
    thread.join()
    print('Main execution finished')

Implementing Thread Communication

Understanding the Global Interpreter Lock

Python's GIL (Global Interpreter Lock) ensures that only one thread executes Python bytecode at a time. While this prevents true parallel execution of CPU-bound tasks, I/O-bound operations can still benefit from threading as threads release the GIL during waiting periods.

Using Thread-Safe Queues

The queue module provides thread-safe queue implementations perfect for inter-thread communication:

from threading import Thread
from queue import Queue
from time import sleep

class DownloadWorker(Thread):
    def __init__(self, task_id, result_queue):
        super(DownloadWorker, self).__init__()
        self.task_id = task_id
        self.result_queue = result_queue

    def download(self, task_id):
        print(f"Downloading task {task_id}...")
        sleep(2)
        return task_id + 100

    def run(self):
        result = self.download(self.task_id)
        self.result_queue.put((self.task_id, result))


class TransformWorker(Thread):
    def __init__(self, input_queue):
        super(TransformWorker, self).__init__()
        self.input_queue = input_queue

    def transform(self, task_id, data):
        print(f"Transforming task {task_id} with data {data}")

    def run(self):
        while True:
            task_id, data = self.input_queue.get()
            if data is None:
                break
            self.transform(task_id, data)

if __name__ == '__main__':
    queue = Queue()
    workers = [DownloadWorker(i, queue) for i in range(1, 11)]
    transformer = TransformWorker(queue)

    for worker in workers:
        worker.start()

    transformer.start()

    for worker in workers:
        worker.join()

    queue.put((-1, None))
    transformer.join()
    print('All tasks completed')

Creating Compressed Tar Archives

The tarfile module handles archive creation with various compression options:

import tarfile
import os

def create_archive(archive_name):
    archive = tarfile.open(archive_name, 'w:gz')

    for filename in os.listdir('.'):
        if filename.endswith('.docx'):
            archive.add(filename)
            os.remove(filename)

    archive.close()
    print(archive.members)

    if not archive.members:
        os.remove(archive_name)

create_archive('Documents.tgz')

Thread Synchronization with Events

The threading.Event class enables threads to signal each other for synchronization purposes:

from threading import Event, Thread

def worker(event):
    print('Worker: waiting for signal')
    event.wait()
    print('Worker: received signal, proceeding')

if __name__ == '__main__':
    signal = Event()
    thread = Thread(target=worker, args=(signal,))
    thread.start()
    signal.set()
    print('Main: signal sent')

The wait() method blocks until another thread calls set(), making this useful for coordinating startup sequecnes or triggering sbusequent operations.