How to use Python as a lazy artifact?

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.

Keywords: PHP

Added by stuart7398 on Fri, 17 Dec 2021 05:55:24 +0200