[11]}ta odpowiedź ma właściwie dwie sugestie - moją pierwszą i drugą, którą odkryłem po pierwszej.

Sched

Podejrzewam, że szukasz sched moduł .

EDIT: moja sugestia wydawała się mało pomocna po przeczytaniu. Postanowiłem więc przetestować moduł sched, aby sprawdzić, czy może on działać zgodnie z moimi sugestiami. Oto mój test: użyłbym go z pojedynczym gwintem, mniej więcej w ten sposób: {]}

class SchedulingThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)
        self.scheduler = sched.scheduler(time.time, time.sleep)
        self.queue = []
        self.queue_lock = threading.Lock()
        self.scheduler.enter(1, 1, self._schedule_in_scheduler, ())

    def run(self):
        self.scheduler.run()

    def schedule(self, function, delay):
        with self.queue_lock:
            self.queue.append((delay, 1, function, ()))

    def _schedule_in_scheduler(self):
        with self.queue_lock:
            for event in self.queue:
                self.scheduler.enter(*event)
                print "Registerd event", event
            self.queue = []
        self.scheduler.enter(1, 1, self._schedule_in_scheduler, ())

Najpierw stworzyłbym Klasa wątku, która miałaby własny scheduler i kolejkę. Co najmniej jedno zdarzenie zostanie zarejestrowane w harmonogramie: jedno do wywołania metody planowania zdarzeń z kolejki.

class SchedulingThread(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.scheduler = sched.scheduler(time.time, time.sleep)
        self.queue = []
        self.queue_lock = threading.Lock()
        self.scheduler.enter(1, 1, self._schedule_in_scheduler, ())

Metoda planowania zdarzeń z kolejki będzie blokować kolejkę, planować każde zdarzenie, opróżniać kolejkę i planować się ponownie, w celu szukania nowych zdarzeń w przyszłości. Pamiętaj, że okres poszukiwania nowych wydarzeń jest krótki( jedna sekunda), możesz zmienić it:

    def _schedule_in_scheduler(self):
        with self.queue_lock:
            for event in self.queue:
                self.scheduler.enter(*event)
                print "Registerd event", event
            self.queue = []
        self.scheduler.enter(1, 1, self._schedule_in_scheduler, ())

Klasa powinna mieć również metodę planowania zdarzeń użytkownika. Oczywiście ta metoda powinna zablokować kolejkę podczas aktualizacji:

    def schedule(self, function, delay):
        with self.queue_lock:
            self.queue.append((delay, 1, function, ()))

Wreszcie, klasa powinna wywołać główną metodę schedulera:

    def run(self):
        self.scheduler.run()

Oto przykład użycia:

def print_time():
    print "scheduled:", time.time()


if __name__ == "__main__":
    st = SchedulingThread()
    st.start()          
    st.schedule(print_time, 10)

    while True:
        print "main thread:", time.time()
        time.sleep(5)

    st.join()

Jego wyjście w mojej maszynie to:

$ python schedthread.py
main thread: 1311089765.77
Registerd event (10, 1, <function print_time at 0x2f4bb0>, ())
main thread: 1311089770.77
main thread: 1311089775.77
scheduled: 1311089776.77
main thread: 1311089780.77
main thread: 1311089785.77

Ten kod to tylko szybki przykład, może wymagać trochę pracy. Muszę jednak przyznać, że jestem trochę zafascynowany modułem sched, więc sugerowałem to. Możesz też poszukać innych propozycji:) [12]}

APScheduler

Szukając w Google rozwiązań takich jak ten, który napisałem, znalazłem ten niesamowity moduł APScheduler . To jest tak praktyczne i użyteczne, że założę się, że jest Twoim rozwiązaniem. Mój poprzedni przykład byłby o wiele prostszy z tym modułem:

from apscheduler.scheduler import Scheduler
import time

sch = Scheduler()
sch.start()

@sch.interval_schedule(seconds=10)

def print_time():
    print "scheduled:", time.time()
    sch.unschedule_func(print_time)

while True:
    print "main thread:", time.time()
    time.sleep(5)

(niestety nie znalazłem sposobu na zaplanowanie zdarzenia do wykonania tylko raz, więc zdarzenie funkcji powinno samo się wyłączyć. I założę się, że da się to rozwiązać jakimś dekoratorem.)

Jeśli masz kilka zadań, które muszą zostać wykonane później, i chcesz, aby utrzymywały się, nawet jeśli zamkniesz program wywołujący lub Twoich pracowników, powinieneś naprawdę spojrzeć na selery , co sprawia, że bardzo łatwo jest tworzyć nowe zadania, wykonywać je na dowolnej maszynie, i czekać na wyniki.

Ze strony selera "jest to proste zadanie dodawanie dwóch liczb:"

from celery.task import task

@task
def add(x, y):
    return x + y

Możesz wykonać zadanie w tle, lub poczekać na zakończenie:

>>> result = add.delay(8, 8)
>>> result.wait() # wait for and return the result
16

Napisałeś:

Jednym z problemów jest to, że proces używa zeromq do komunikacji, więc potrzebuję trochę integracji (eventlet już to ma)

Wydaje się, że twój wybór będzie miał duży wpływ na te szczegóły, które są nieco niejasne-w jaki sposób zeromq jest używany do komunikacji, ile zasobów będzie wymagać integracja, a jakie są Twoje wymagania i dostępne zasoby.

Używałem selera dla kilku projektów (te są hostowane na ep.io PaaS hosting, który zapewnia łatwy sposób korzystania z niego).

Seler wygląda na dość elastyczne rozwiązanie, pozwalające opóźniać zadania, oddzwaniać, wygaśnięcie i ponowna próba zadania, ograniczenie szybkości wykonywania zadań itp. Może być używany z Redis, Beanstalk, CouchDB, MongoDB lub bazą danych SQL.

Przykładowy kod (definicja zadania i asynchroniczne wykonanie po opóźnieniu):

from celery.decorators import task

@task
def my_task(arg1, arg2):
    pass # Do something

result = my_task.apply_async(
    args=[sth1, sth2], # Arguments that will be passed to `my_task()` function.
    countdown=3, # Time in seconds to wait before queueing the task.
)

Zobacz także sekcję w seler docs .

Spojrzałeś na moduł multiprocessing? Jest standardowo wyposażony w Python. Jest podobny do modułu threading, ale uruchamia każde zadanie w procesie. Można użyć obiektu Pool() do skonfigurowania puli roboczej, a następnie użyć metody .map() do wywołania funkcji z różnymi kolejkowymi argumentami zadania.

Pyzmq posiada implementację ioloop z podobnym api jak tornado ioloop. Realizuje on DelayedCallback co może Ci pomóc.

Zakładając, że twój proces ma pętlę run, która może odbierać sygnały, a czas każdej akcji mieści się w granicach operacji sekwencyjnych, użyj sygnałów i POSIX alarm ()

    signal.alarm(time)
If time is non-zero, this function requests that a 
SIGALRM signal be sent to the process in time seconds. 

To zależy od tego, co rozumiesz przez "te" późniejsze "działania mogą być dużo" i czy twój proces już używa sygnałów. Ze względu na sformułowanie pytania nie jest jasne, dlaczego zewnętrzny pakiet Pythona byłby potrzebny.

Inną opcją jest użycie wiązań Phyton GLib , w szczególności jego funkcji timeout.

Jest to dobry wybór, o ile nie chcesz używać wielu rdzeni i o ile zależność od GLib nie stanowi problemu. Obsługuje wszystkie zdarzenia w tym samym wątku, co zapobiega problemom z synchronizacją. Dodatkowo jego ramka zdarzeń może być również używana do obserwacji i obsługi zdarzeń opartych na IO (np. socketach).

UPDATE:

Oto sesja na żywo z wykorzystaniem GLib: {]}

>>> import time
>>> import glib
>>> 
>>> def workon(thing):
...     print("%s: working on %s" % (time.time(), thing))
...     return True # use True for repetitive and False for one-time tasks
... 
>>> ml = glib.MainLoop()
>>> 
>>> glib.timeout_add(1000, workon, "this")
2
>>> glib.timeout_add(2000, workon, "that")
3
>>> 
>>> ml.run()
1311343177.61: working on this
1311343178.61: working on that
1311343178.61: working on this
1311343179.61: working on this
1311343180.61: working on this
1311343180.61: working on that
1311343181.61: working on this
1311343182.61: working on this
1311343182.61: working on that
1311343183.61: working on this

Moim zdaniem przydałoby się coś o nazwie "multitasking kooperacyjny". To pokręcona rzecz i jest naprawdę fajna. Wystarczy spojrzeć na prezentację PyCon z 2010 roku: http://blip.tv/pycon-us-videos-2009-2010-2011/pycon-2010-cooperative-multitasking-with-twisted-getting-things-done-concurrently-11-3352182

Cóż, będziesz potrzebował kolejki transportowej, aby to zrobić...

Proste. Możesz dziedziczyć swoją klasę z wątku i tworzyć instancję swojej klasy z param jak timeout, więc dla każdej instancji twojej klasy możesz powiedzieć timeout, który sprawi, że Twój wątek będzie czekał na ten czas