Multi Proses dengan Python

Berawal dari server yang bengong karena proses for-loop yang berkepanjangan, saatnya untuk eksplorasi multi proses di Python. Artinya saat looping, proses akan dikerjakan oleh ‘worker’, sehingga loop bisa berlanjut ke baris berikutnya. Cocok untuk pemrosesan data per baris, seperti query lalu update satu-satu dengan proses panjang, atau proses unggah berkas per baris untuk satu persatu dimasukkan dalam database.

Ada beberapa alternatif:

1. Thread + Queue
   Ini standar bawaan Python. Langsung bisa pakai dengan kinerja yang mengagumkan. Kelemahannya adalah saat mengerjakan sesuatu tidak bisa ambil nilai kembalian hasil kerjaan tersebut (return value), kecuali lewat sub-classing dengan mengubah perilaku kelas utama dari Thread.
2. PythonRQ + Redis
   Antriannya banyak pilihan dan bisa ambil nilai kembalian setelah menerapkan delay (1 detik). PythonRQ menerapkan fork untuk workernya (bukan thread), sehingga setiap proses berdiri sendiri. Tergantung sama Redis untuk menyimpan data antriannya. Cocok untuk distribusi kerjaan, tapi untuk kecepatan, kurang.
3. ThreadPool
   Merupakan sub modul dari modul multiprocessing. Lebih cepat dari PythonRQ namun lebih lambat dari native Thread+Queue. Kelebihannya adalah bisa ambil nilai kembalian dengan mudah tanpa harus sub-classing.

Kode

Berikut contoh kode yang saya buat untuk pengujian:

import os
import sys, time
from Queue import Queue

from threading import Thread
from rq import Queue as rqueue
from redis import Redis

q = Queue()
threads = 10
numrange = 1000

# fungsi yang akan dijalankan
def deltrans(i):
    return "hello %s"%(i)

#THREAD + QUEUE
start = time.time()

def dothing(q):
    while True:
        res = q.get()
        q.task_done
    return res

for y in range(threads):
    worker = Thread(target=dothing, args =(q,))
    worker.setDaemon(True)
    worker.start()

for i in range(numrange):
    q.put(deltrans(i))
qtime = time.time()-start

#PythonRQ
astart = time.time()
conn = Redis()

rq = rqueue(connection = conn)

from hworld import helloworld

for i in range(numrange):
    hw = rq.enqueue(helloworld,i)
    print hw.result

# ThreadPool
btime = time.time()-astart
from multiprocessing.pool import ThreadPool

pool = ThreadPool(processes=threads)
astart = time.time()

for i in range(numrange):
    async_result = pool.apply_async(deltrans, (i,)) 
    print async_result.get()

ctime = time.time()-astart

print "%s %s %s"%(qtime,btime,ctime)

Tambahan berkas hworld.py untuk PythonRQ:

def helloworld(i):
    return "hello %s"%(i)

Hasil

1. Thread+ Queue: 0.0666551589966 (tanpa kembalian)
2. PythonRQ: 0.510520935059 (tanpa kembalian, karena harus tunggu 1 detik untuk ambil kembalian)
3. ThreadPool: 0.222325086594 (dengan kembalian)

Pendapat pribadi, jika tidak ingin kembalian, gunakan Thread+Queue. Jika ingin kembalian, gunakan ThreadPool. Jika ingin banyak manipulasi kembalian, gunakan PythonRQ.