Multiprocessing vs Multithreading di Python: Panduan Lengkap

Concurrency dan parallelism adalah konsep penting dalam programming modern untuk meningkatkan performa aplikasi. Python menyediakan dua pendekatan utama: multithreading dan multiprocessing. Dalam tutorial ini, kita akan mempelajari perbedaan keduanya, kapan menggunakan masing-masing, dan bagaimana mengimplementasikannya dengan efektif.

Konsep Dasar

Threading

• Lightweight (ringan)
• Berbagi memory space
• Context switching lebih cepat
• Cocok untuk I/O-bound tasks

Multiprocessing

• Heavyweight (lebih berat)
• Memory space terpisah
• True parallelism
• Cocok untuk CPU-bound tasks

Global Interpreter Lock (GIL)

Apa itu GIL?

GIL adalah mutex (mutual exclusion lock) yang melindungi akses ke Python objects, mencegah multiple threads mengeksekusi Python bytecode secara bersamaan.

import threading
import time

CPU-bound task
def cpuintensive():

    count = 0
    for i in range(50000000):

        count += 1
    return count

Single thread
start = time.time()
cpuintensive()
cpuintensive()

end = time.time()
print(f"Single thread: {end - start:.2f}s")

Multi-threading (TIDAK lebih cepat karena GIL!)
start = time.time()
t1 = threading.Thread(target=cpuintensive)
t2 = threading.Thread(target=cpuintensive)

t1.start()
t2.start()
t1.join()
t2.join()
end = time.time()
print(f"Multi-threading: {end - start:.2f}s")

Single thread: 3.45s
Multi-threading: 3.52s  # Tidak ada speedup!

import threading
import requests
import time

urls = [
    'https://jsonplaceholder.typicode.com/posts/1',
    'https://jsonplaceholder.typicode.com/posts/2',
    'https://jsonplaceholder.typicode.com/posts/3',
    'https://jsonplaceholder.typicode.com/posts/4',
    'https://jsonplaceholder.typicode.com/posts/5',
]

def download(url):
    """Download content dari URL"""
    response = requests.get(url)
    print(f"Downloaded {url}: {len(response.content)} bytes")

Sequential (tanpa threading)
start = time.time()
for url in urls:
    download(url)
sequentialtime = time.time() - start
print(f"\nSequential: {sequentialtime:.2f}s")


Multi-threading
start = time.time()
threads = []
for url in urls:
    thread = threading.Thread(target=download, args=(url,))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

threadedtime = time.time() - start
print(f"Multi-threading: {threadedtime:.2f}s")

print(f"Speedup: {sequentialtime/threadedtime:.2f}x")

Downloaded https://jsonplaceholder.typicode.com/posts/1: 292 bytes
Downloaded https://jsonplaceholder.typicode.com/posts/2: 293 bytes
...

Sequential: 2.45s
Multi-threading: 0.62s
Speedup: 3.95x  # Significant speedup!

import threading
import sqlite3
import time

def querydatabase(query, resultlist, index):

    """Execute database query"""
    conn = sqlite3.connect('example.db')
    cursor = conn.cursor()

    cursor.execute(query)
    result = cursor.fetchall()
    resultlist[index] = result

    conn.close()
    print(f"Query {index+1} completed: {len(result)} rows")

Multiple queries
queries = [
    "SELECT  FROM users WHERE age > 25",

    "SELECT  FROM orders WHERE total > 1000",
    "SELECT  FROM products WHERE stock < 10",

    "SELECT  FROM customers WHERE country = 'Indonesia'"
]

results = [None]  len(queries)


With threading
start = time.time()
threads = []
for i, query in enumerate(queries):
    thread = threading.Thread(target=querydatabase, args=(query, results, i))

    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

print(f"All queries completed in {time.time() - start:.2f}s")

import multiprocessing
import time
import math

def cpuintensivetask(n):

    """CPU-intensive calculation"""
    count = 0
    for i in range(n):
        count += math.sqrt(i)  math.sin(i)
    return count

Parameters
numbers = [10000000, 10000000, 10000000, 10000000]


Sequential
start = time.time()
results = [cpuintensivetask(n) for n in numbers]

sequentialtime = time.time() - start
print(f"Sequential: {sequentialtime:.2f}s")


Multiprocessing
start = time.time()
with multiprocessing.Pool(processes=4) as pool:
    results = pool.map(cpuintensivetask, numbers)

multiprocessingtime = time.time() - start
print(f"Multiprocessing: {multiprocessingtime:.2f}s")

print(f"Speedup: {sequentialtime/multiprocessingtime:.2f}x")

Sequential: 12.45s
Multiprocessing: 3.28s
Speedup: 3.80x  # Nearly linear speedup!

import multiprocessing
from PIL import Image
import os

def processimage(imagepath):

    """Apply filters to image"""
    img = Image.open(imagepath)

    # Convert to grayscale
    img = img.convert('L')

    # Resize
    img = img.resize((800, 600))

    # Save
    outputpath = f"processed{os.path.basename(imagepath)}"

    img.save(outputpath)

    return outputpath


List of images
imagefiles = [
    'image1.jpg',
    'image2.jpg',
    'image3.jpg',
    'image4.jpg',
    'image5.jpg',
    'image6.jpg',
]

Sequential processing
start = time.time()
for imgfile in imagefiles:
    processimage(imgfile)
sequentialtime = time.time() - start

print(f"Sequential: {sequentialtime:.2f}s")

Parallel processing
start = time.time()
with multiprocessing.Pool(processes=multiprocessing.cpucount()) as pool:

    results = pool.map(processimage, imagefiles)

paralleltime = time.time() - start
print(f"Parallel: {paralleltime:.2f}s")

print(f"Speedup: {sequentialtime/paralleltime:.2f}x")

from concurrent.futures import ThreadPoolExecutor
import time

def task(name, delay):
    print(f"Task {name} starting")
    time.sleep(delay)
    print(f"Task {name} completed")
    return f"Result from {name}"

Using ThreadPoolExecutor
with ThreadPoolExecutor(maxworkers=3) as executor:
    # Submit tasks
    future1 = executor.submit(task, "A", 2)
    future2 = executor.submit(task, "B", 1)
    future3 = executor.submit(task, "C", 3)

    # Get results
    print(future1.result())  # Blocks until complete
    print(future2.result())
    print(future3.result())

Map with ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor
import requests

def fetchurl(url):

    response = requests.get(url)
    return len(response.content)

urls = [
    'https://example.com',
    'https://python.org',
    'https://github.com',
]

Using map
with ThreadPoolExecutor(maxworkers=5) as executor:
    results = executor.map(fetchurl, urls)


    for url, size in zip(urls, results):
        print(f"{url}: {size} bytes")

from concurrent.futures import ThreadPoolExecutor, ascompleted
import time
import random

def task(n):
    sleeptime = random.uniform(0.5, 2.0)

    time.sleep(sleeptime)
    return f"Task {n} completed in {sleeptime:.2f}s"


tasks = list(range(10))

with ThreadPoolExecutor(maxworkers=3) as executor:
    # Submit all tasks
    futuretotask = {executor.submit(task, t): t for t in tasks}

    # Process as they complete
    for future in ascompleted(futuretotask):

        tasknum = futuretotask[future]
        try:
            result = future.result()
            print(result)
        except Exception as e:
            print(f"Task {tasknum} generated exception: {e}")

from concurrent.futures import ProcessPoolExecutor
import os

def cputask(n):
    """CPU-intensive task"""
    print(f"Process {os.getpid()} working on {n}")
    result = sum(ii for i in range(n))

    return result

numbers = [1000000, 2000000, 3000000, 4000000]

with ProcessPoolExecutor(maxworkers=4) as executor:

    results = executor.map(cputask, numbers)

    for num, result in zip(numbers, results):
        print(f"Sum of squares up to {num}: {result}")

from multiprocessing import Process, Queue, Manager
import time

Using Queue
def producer(queue):
    for i in range(5):
        queue.put(f"Item {i}")
        print(f"Produced: Item {i}")
        time.sleep(0.5)

def consumer(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        print(f"Consumed: {item}")

Create queue
queue = Queue()

Create processes
p1 = Process(target=producer, args=(queue,))
p2 = Process(target=consumer, args=(queue,))

p1.start()
p2.start()

p1.join()
queue.put(None)  # Signal to stop consumer
p2.join()

Using Manager (for shared state)
def worker(shareddict, key, value):

    shareddict[key] = value
    print(f"Set {key} = {value}")

manager = Manager()
shareddict = manager.dict()


processes = []
for i in range(5):
    p = Process(target=worker, args=(shareddict, f'key{i}', i10))
    processes.append(p)
    p.start()

for p in processes:
    p.join()

print("Final shared dict:", dict(shareddict))

from multiprocessing import Pool
from tqdm import tqdm
import time

def slowtask(x):
    time.sleep(0.5)  # Simulate work
    return x  x


numbers = range(20)

With progress bar
with Pool(processes=4) as pool:
    results = list(tqdm(
        pool.imap(slowtask, numbers),

        total=len(numbers),
        desc="Processing"
    ))

print(f"Results: {results}")

import threading

Without lock (race condition)
counter = 0

def increment():
    global counter
    for  in range(100000):
        counter += 1

threads = [threading.Thread(target=increment) for  in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Counter without lock: {counter}")  # Incorrect! < 1000000

With lock
counter = 0
lock = threading.Lock()

def incrementwithlock():

    global counter
    for  in range(100000):
        with lock:  # or lock.acquire() ... lock.release()
            counter += 1

threads = [threading.Thread(target=incrementwithlock) for  in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Counter with lock: {counter}")  # Correct! = 1000000

import threading
import time

Limit concurrent access to resource
semaphore = threading.Semaphore(3)  # Max 3 threads

def accessresource(threadid):

    print(f"Thread {threadid} waiting...")
    semaphore.acquire()
    try:
        print(f"Thread {threadid} accessing resource")

        time.sleep(2)  # Simulate work
    finally:
        print(f"Thread {threadid} releasing resource")
        semaphore.release()

threads = [threading.Thread(target=accessresource, args=(i,)) for i in range(10)]

for t in threads:
    t.start()
for t in threads:
    t.join()

import threading
import time

event = threading.Event()

def waiter():
    print("Waiting for event...")
    event.wait()  # Block until set
    print("Event received! Proceeding...")

def setter():
    print("Working...")
    time.sleep(3)
    print("Setting event")
    event.set()

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()

t1.join()
t2.join()

import time
import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

I/O-bound task
def iotask(n):
    time.sleep(0.1)  # Simulate I/O
    return n  2

CPU-bound task
def cputask(n):

    return sum(ii for i in range(n))


Test data
iodata = list(range(20))
cpudata = [1000000]  8

print("=== I/O-bound Tasks ===")

Sequential I/O
start = time.time()
results = [iotask(x) for x in iodata]

print(f"Sequential: {time.time() - start:.2f}s")

Threading I/O
start = time.time()
with ThreadPoolExecutor(maxworkers=10) as executor:
    results = list(executor.map(iotask, iodata))
print(f"Threading: {time.time() - start:.2f}s")

Multiprocessing I/O (overhead!)
start = time.time()
with ProcessPoolExecutor(maxworkers=4) as executor:

    results = list(executor.map(iotask, iodata))

print(f"Multiprocessing: {time.time() - start:.2f}s")

print("\n=== CPU-bound Tasks ===")

Sequential CPU
start = time.time()
results = [cputask(x) for x in cpudata]

print(f"Sequential: {time.time() - start:.2f}s")

Threading CPU (no speedup!)
start = time.time()
with ThreadPoolExecutor(maxworkers=4) as executor:
    results = list(executor.map(cputask, cpudata))
print(f"Threading: {time.time() - start:.2f}s")

Multiprocessing CPU
start = time.time()
with ProcessPoolExecutor(maxworkers=4) as executor:

    results = list(executor.map(cputask, cpudata))

print(f"Multiprocessing: {time.time() - start:.2f}s")

=== I/O-bound Tasks ===
Sequential: 2.01s
Threading: 0.21s      # 9.5x faster!
Multiprocessing: 0.45s  # Overhead

=== CPU-bound Tasks ===
Sequential: 3.42s
Threading: 3.51s      # No speedup (GIL)
Multiprocessing: 0.94s  # 3.6x faster!

# Decision tree
def chooseconcurrency(tasktype):

    if tasktype == "I/O-bound":
        return "Use Threading or AsyncIO"
    elif tasktype == "CPU-bound":

        return "Use Multiprocessing"
    else:
        return "Profile first!"

Examples:
- Web scraping → Threading
- API calls → Threading
- File I/O → Threading
- Image processing → Multiprocessing
- Data analysis → Multiprocessing
- Scientific computing → Multiprocessing

import os

For I/O-bound (threading)
ioworkers = 10  # Or more, depends on I/O latency

For CPU-bound (multiprocessing)
cpuworkers = os.cpucount()  # Number of CPU cores

Conservative approach
cpuworkers = os.cpucount() - 1  # Leave one core free

3. Handle Errors Gracefully

from concurrent.futures import ThreadPoolExecutor, ascompleted

def riskytask(n):
    if n == 5:
        raise ValueError("Number 5 is not allowed!")
    return n  2


numbers = range(10)

with ThreadPoolExecutor(maxworkers=3) as executor:

    futuretonum = {executor.submit(riskytask, n): n for n in numbers}

    for future in ascompleted(futuretonum):

        num = futuretonum[future]

        try:
            result = future.result()
            print(f"{num} -> {result}")
        except Exception as e:
            print(f"{num} generated exception: {e}")

from concurrent.futures import ThreadPoolExecutor
import requests

def fetchwithcleanup(url):

    session = requests.Session()
    try:
        response = session.get(url)
        return response.statuscode
    finally:
        session.close()  # Cleanup

urls = ['https://example.com']  10

with ThreadPoolExecutor(maxworkers=5) as executor:

    results = executor.map(fetchwithcleanup, urls)

    print(list(results))
Executor automatically calls shutdown()

import threading

❌ Bad: Race condition
sharedlist = []

def appenditems():

    for i in range(1000):
        sharedlist.append(i)

threads = [threading.Thread(target=appenditems) for  in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(len(sharedlist))  # May not be 10000!


✅ Good: Use lock or thread-safe structure
from queue import Queue

sharedqueue = Queue()

def appenditemssafe():
    for i in range(1000):
        sharedqueue.put(i)


threads = [threading.Thread(target=appenditemssafe) for  in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(sharedqueue.qsize())  # Exactly 10000

import threading

lock1 = threading.Lock()
lock2 = threading.Lock()

❌ Bad: Potential deadlock
def thread1():
    lock1.acquire()
    lock2.acquire()  # May wait forever if thread2 has lock2
    # ... work ...
    lock2.release()
    lock1.release()

def thread2():
    lock2.acquire()
    lock1.acquire()  # May wait forever if thread1 has lock1
    # ... work ...
    lock1.release()
    lock2.release()

✅ Good: Consistent lock ordering
def thread1safe():
    lock1.acquire()
    lock2.acquire()
    # ... work ...
    lock2.release()
    lock1.release()

def thread2_safe():
    lock1.acquire()  # Same order!
    lock2.acquire()
    # ... work ...
    lock2.release()
    lock1.release()

3. Pickling Issues (Multiprocessing)

from multiprocessing import Pool

❌ Bad: Lambda functions can't be pickled
numbers = [1, 2, 3, 4, 5]
pool.map(lambda x: x2, numbers)  # Error!


✅ Good: Use regular function
def double(x):
    return x  2

with Pool(processes=4) as pool:
    results = pool.map(double, numbers)

Kesimpulan

Quick Reference

| Aspect | Threading | Multiprocessing |

|--------|-----------|-----------------|

| Best for | I/O-bound tasks | CPU-bound tasks |

| GIL Impact | Limited by GIL | Bypasses GIL |

| Memory | Shared | Separate |

| Overhead | Low | High |

| Communication | Easy (shared memory) | Complex (IPC) |

| Debugging | Easier | Harder |

Decision Matrix

• Network I/O (API calls, web scraping)
• File I/O
• Database queries
• Waiting for external resources

• Heavy computation
• Data processing
• Image/video processing
• Scientific computing
• Embarrassingly parallel problems

• High concurrency I/O
• Web servers
• Real-time applications
• (Tutorial terpisah untuk AsyncIO)

Key Takeaways

Selamat menggunakan concurrency di Python! 🚀