In our daily work, we often use tasks that need to be executed periodically. One way is to use the crond[1] of Linux system combined with the command line. Another way is to use Python directly. What follows is the implementation of common Python timing tasks.
1. Using while True: + sleep() to implement scheduled tasks
The sleep(secs) function in the time module can make the currently executing thread pause for secs seconds before continuing execution. The so-called pause means that the current thread enters the blocking state. When the time specified by the sleep() function is reached, it will change from the blocking state to the ready state and wait for CPU scheduling.
Based on this feature, we can implement simple timing tasks through while loop + sleep().
Code example:
import datetime import time def time_printer(): now = datetime.datetime.now() ts = now.strftime('%Y-%m-%d %H:%M:%S') print('do func time :', ts) def loop_monitor(): while True: time_printer() time.sleep(5) # Pause for 5 seconds if __name__ == "__main__": loop_monitor()
Main disadvantages:
- Only the interval can be set, and the specific time cannot be specified, such as 8:00 every morning
- Sleep is a blocking function, that is, during sleep, the program can't operate at all.
2. Running scheduled tasks using the Timeloop Library
Timeloop[2] is a library that can be used to run multi cycle tasks. This is a simple library that uses decorator mode to run marker functions in threads.
Example code:
import time from timeloop import Timeloop from datetime import timedelta tl = Timeloop() @tl.job(interval=timedelta(seconds=2)) def sample_job_every_2s(): print "2s job current time : {}".format(time.ctime()) @tl.job(interval=timedelta(seconds=5)) def sample_job_every_5s(): print "5s job current time : {}".format(time.ctime()) @tl.job(interval=timedelta(seconds=10)) def sample_job_every_10s(): print "10s job current time : {}".format(time.ctime())
3. Using threading Timer implements scheduled tasks
Timer in threading module is a non blocking function, which is a little better than sleep. Timer is basically understood as timer. We can start multiple timer tasks. These timer tasks are executed asynchronously, so there is no waiting order execution problem.
Timer(interval, function, args=[ ], kwargs={ })
- interval: specified time
- function: method to execute
- args/kwargs: parameters of method
Code example:
import datetime from threading import Timer def time_printer(): now = datetime.datetime.now() ts = now.strftime('%Y-%m-%d %H:%M:%S') print('do func time :', ts) loop_monitor() def loop_monitor(): t = Timer(5, time_printer) t.start() if __name__ == "__main__": loop_monitor()
Note: Timer can only be executed once, which needs to be called circularly, otherwise it can only be executed once
4. Using the built-in module sched to realize scheduled tasks
sched module implements a general event scheduler, which uses a delay function in the scheduler class to wait for a specific time to execute tasks. It also supports multi-threaded applications. After each task is executed, the delay function will be called immediately to ensure that other threads can also execute.
class sched. The scheduler (timefunc, delayfunc) class defines a general interface for scheduling events. It requires two external parameters. Timefunc is a function that returns time type numbers without parameters (commonly used, such as time in the time module). Delayfunc should be a function that requires a parameter to call and is compatible with the output of timefunc It also acts as a function of delaying multiple time units (commonly used, such as sleep in the time module).
Code example:
import datetime import time import sched def time_printer(): now = datetime.datetime.now() ts = now.strftime('%Y-%m-%d %H:%M:%S') print('do func time :', ts) loop_monitor() def loop_monitor(): s = sched.scheduler(time.time, time.sleep) # Build scheduler s.enter(5, 1, time_printer, ()) s.run() if __name__ == "__main__": loop_monitor()
Main methods of scheduler object:
- enter(delay, priority, action, argument) to schedule an event to delay the delay by a time unit.
- cancel(event): deletes an event from the queue. If the event is not in the current queue, the method will run a ValueError.
- run(): run all scheduled events. This function will wait (using the delayfunc() function passed to the constructor) and execute the event until there are no more scheduled events.
Personal comments: better than threading Timer is better and does not need to be called in a loop.
Python is a very diverse and well-developed language, so there will certainly be many functions I haven't considered. If you know anything, you can tell me in the comment area. If you see here, please point a praise before you go ~ thank you for reading.