Job Queue mit RabbitMQ und Pika (oder Alternativen?)

[SimonR]

Moin zusammen!

Ich möchte eine Warteschlange für rechenintensive Jobs in Python realisieren. Ich habe bereits ein funktionierendes Grundgerüst, würde jedoch gerne weitere Funktionen unterbringen.

Eine kurze Erklärung:
Die übergebenen Pfade sind Netzwerkpfade und auf jedem Rechner der Jobs annimmt gleich. Das Hauptproblem ist, dass die Software die die Rechnungen ausführt Schreibzugriff auf sein Installationsverzeichnis benötigt und jede Installation nur einen Job abarbeiten kann. Deshalb hat jeder Rechner eine Programminstallation pro CPU-Kern und eine entsprechend angepasste executable Variable (sonst könnte ich wohl z.B. die Sun Grid Engine o.ä. nutzen).

Meine Frage ist nun:
Ich würde gerne Jobs löschen und Job-Prioritäten festlegen können. Habt Ihr eine Idee, wie man das am geschicktesten realisieren kann? Ich bin jetzt nicht auf Code-Schnipsel aus, sondern hoffe lediglich auf Denkanstöße.
Habt Ihr schonmal Warteschlangen in Python programmiert? Wenn ja, worauf basieren diese? Gibt es gute Alternativen zu RabbitMQ / Pika. Via Suchmaschine bin ich nicht fündig geworden (vielleicht hab ich auch nur die falschen Schlagworte verwendet; bitte entschuldigt falls ich da was Offensichtliches nicht gesehen habe)

Vielen Dank und viele Grüße
Simon

Nachtrag:
Angenommen das Ganze findet auf einem einzigen Rechner mit 16 Kernen statt (ich höre grad, dass evtl. einer angeschafft wird ), wie würdet Ihr dann an das Problem rangehen?
Das Threading Modul ist doch zum aufteilen von Jobs auf mehrere Kerne und nicht zur Verteilung/Verwaltung von Jobs? Oder versteh ich das jetzt falsch?

submit.py:

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
import os
import pika

if len(sys.argv) <= 1:
    sys.exit()

path = os.getcwd()

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.166'))
channel = connection.channel()

channel.queue_declare(queue='job_queue', durable=True)

job = os.path.join(path, sys.argv[1])

channel.basic_publish(exchange='',
                      routing_key='jobQueue',
                      body=job,
                      properties=pika.BasicProperties(
                          delivery_mode = 2, # make message persistent
                      ))
print 'Job submitted {0}'.format(job)
connection.close()

worker.py (hier vereinfacht)

Code: Alles auswählen

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import pika
from subprocess import call

executable = 'c:\\blubb2\\bla.exe'

def callback(ch, method, properties, body):
    print 'Received {1}'.format(body)
    execute = executable, body, '-someParameter comeValue'
    call(execute)
    ch.basic_ack(delivery_tag = method.delivery_tag)
   

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.0.166'))
channel = connection.channel()
channel.queue_declare(queue='jobQueue', durable=True)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue='jobQueue')

channel.start_consuming()