Code: Alles auswählen
class myThread (threading.Thread):
def __init__(self, name, parameter):
threading.Thread.__init__(self)
self.name = name
self.para = parameter
def run(self):
print "Starting " + self.name
funktion(self.para)
print "Exiting " + self.name
Liste = ["element", "element2", ...]
for i in Liste:
myThread("Thread_"+i, i).start()
Code: Alles auswählen
from concurrent.futures import ThreadPoolExecutor
def funktion(_argument):
pass
def main():
liste = ['element', 'element2']
executor = ThreadPoolExecutor(2)
for future in executor.map(funktion, liste):
future.result()
Habe einige Dinge berücksichtigt, die in der "quick 'n' dirty" Fassung wirklich nicht gut durchdacht wurden: siehe PastebinSirius3 hat geschrieben:@droptix: die netten Strings, die Du überall zwischen Deine Befehle streust, erinnern irgendwie an doc-Strings, sind aber keine und sollten deshalb auch richtige Kommentare werden. Die Namen halten sich nicht an PEP8.
- keine Deiner Klassenvariablen sollte ein Klassenvariable sein. Zwei »QueueWorkers«-Instanzen kommen so auf jeden Fall nicht miteinander klar, wenn sie verschiedene Arbeit erledigen sollen erst recht nicht.
- Du hast wunderschöne busy-Loops implementiert, was man ja mit einer Queue gerade verhindern will.
- Dein »except« sollte nur ein »Empty« abfangen.
- den Fall, dass in »work« eine »Exception« auftritt, ignorierst Du völlig.
- der »_threadLock« ist überflüssig, da Queues an sich threadsicher sind. Solltest Du doch mal einen Lock brauchen, benutze »with«.
- »create« hat einen Parameter, der nicht benutzt wird.
Dann würde ich der QueueWorkers-Klasse noch ein schönes input-output-Interface gönnen, das direkte Ansprechen von inQ und outQ ist unschön. Die work-Funktion sollte man auch von außen festlegen können, z.B. als Parameter in __init__.
Indem du blockierst, wenn _inQ keine Nachrichten enthält (gib dem mal einen richtigen Namen). Du brauchst dann nur noch eine Nachricht für "finishAllTasks". Das ``else: continue`` kannst du dir in der Schleife übrigens sparen.droptix hat geschrieben:busy-Loops: wie verbessern? Was schlägst du vor?
Dein try/except-Block behandelt alle Fehler gleich. Wenn ein NameError in dem try-Block auftritt, dann wirst du den Fehler nie bemerken, bzw. schwer finden. Dir wurde doch bereits der Hinweis zu der Empty-Exception gegeben, lese doch einfach mal die Dokumentation zum Queue-Moduldroptix hat geschrieben:Exceptions: konkrete Tipps (Code-Snippets)?
Genau dafür ist das Queue-Modul da. Wie ich oben schon schrieb: wenn du ein Modul verwendest, dann lese dazu auch die Dokumentation.droptix hat geschrieben:threadLock: bist du da sicher? Du meinst für `Queue.put()` muss man nicht vorher `threading.Lock.aquire()` und danach `...release()` machen?
Code: Alles auswählen
D:
dir D:\Temp
C:\Python27\python.exe D:\Temp\dist\futures\setup.py build
C:\Python27\python.exe D:\Temp\dist\futures\setup.py install
pauseCode: Alles auswählen
import time
import concurrent.futures
# list of tasks: lower-case alphabet letters
tasks = []
for i in range(26):
tasks.append(chr(97+i))
results = []
# what to be done: transform to upper-case
def work(task):
time.sleep(0.1)
results.append(task.upper())
executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
for task in tasks:
future = executor.submit(work, task)
""" It is not blocking here, are the pending tasks processed anyway? """
print "results 1:", results
""" result 1 is incomplete... if we wait a moment, it will be completed """
time.sleep(5)
print "results 2", resultsCode: Alles auswählen
import time
import concurrent.futures
# list of tasks: lower-case alphabet letters
tasks = []
for i in range(26):
tasks.append(chr(97+i))
results = []
# what to be done: transform to upper-case
def work(task):
time.sleep(0.1)
results.append(task.upper())
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
for task in tasks:
future = executor.submit(work, task)
""" It is blocking here, so the result is already complete when printed. """
print "results:", resultsCode: Alles auswählen
#!/usr/bin/env python
from random import random
from string import lowercase, uppercase
from time import sleep
from concurrent.futures import ThreadPoolExecutor
def work(character):
sleep(random())
print character
return character.upper()
def main():
tasks = lowercase
with ThreadPoolExecutor(4) as executor:
results = ''.join(executor.map(work, tasks))
print results, results == uppercase
if __name__ == '__main__':
main()Code: Alles auswählen
with ThreadPoolExecutor(4) as executor:
executor.submit(work, task)
executor.shutdown()Außerdem sind alle `submit()`-Aufrufe ja bereits durchgeführt worden. Ich finde also kein sinnvolles praktisches Beispiel zur Nutzung von `shutdown()`.Doku hat geschrieben:You can avoid having to call this method explicitly if you use the with statement, which will shutdown the Executor (waiting as if Executor.shutdown() were called with wait set to True)
Du meinst, wenn ich das so mache, kann ich auch gleich `map()` nehmen, richtig?BlackJack hat geschrieben:Wenn Du merkst, dass Du mit `submit()` selbst `map()` implementierst, dann solltest Du lieber `map()` nehmen. Man hat nichts mit den `Future`-Exemplaren zu tun und die Reihenfolge bleibt erhalten.
Code: Alles auswählen
tasks = "".join([chr(97+i) for i in range(26)]) # string.lowercase returns more than 26 alphabet characters (Win7x64)
results = []
def work(character):
return character.upper()
with concurrent.futures.ThreadPoolExecutor(4) as executor:
for character in tasks:
future = executor.submit(work, character)
results.append(future.result())
print resultsKonkretes Beispiel für mein Python-Programm:BlackJack hat geschrieben:Warum willst Du mehrere `ThreadPoolExecutor`-Exemplare hintereinander schalten? Statt Funktionen zu kombinieren und „hintereinander zu schalten” und die von *einem* Executor ausführen zu lassen. Das ist wesentlich einfacher. Wenn mehrere parallel laufen sollen und Du die erledigten Aufgaben weitergeben möchtest, dann müsstest Du den `Future`-Exemplaren Rückruffunktionen mitgeben, die dann den nächsten Schritt anstossen.
Code: Alles auswählen
tasks = string.ascii_lowercase
def work(character):
return character.upper()
with concurrent.futures.ThreadPoolExecutor(4) as executor:
futures = [executor.submit(work, character) for character in tasks]
results = [future.result() for future in futures]
#
# oder wenn die Reihenfolge egal ist:
#
results = [
future.result()
for future in concurrent.futures.as_completed(futures)
]
print resultsCode: Alles auswählen
for task in iter_endless_tasks():
executor.submit(work, task).add_done_callback(work_done)Code: Alles auswählen
#!/usr/bin/env python
from __future__ import print_function
from concurrent.futures import ThreadPoolExecutor
from random import random
from threading import Lock
from time import sleep
class ProcessingStep(object):
def __init__(self, work_callable, worker_count, next_step=None):
self.work_callable = work_callable
self.executor = ThreadPoolExecutor(worker_count)
self.next_step = next_step
self.lock = Lock()
def __enter__(self):
return self
def __exit__(self, _type, _value, _traceback):
self.wait()
def submit(self, *arguments):
(
self.executor
.submit(self.work_callable, *arguments)
.add_done_callback(self.work_done)
)
def work_done(self, future):
result = future.result()
if self.next_step:
with self.lock:
self.next_step.submit(result)
def wait(self):
self.executor.shutdown()
if self.next_step:
self.next_step.wait()
def work_a(argument):
sleep(random())
return argument.upper()
def work_b(argument):
sleep(random())
return ' '.join(argument)
def main():
with ProcessingStep(
work_a, 2, ProcessingStep(work_b, 2, ProcessingStep(print, 1))
) as processor:
for task in iter(raw_input, ''):
processor.submit(task)
if __name__ == '__main__':
main()