thread_worker

thread_worker

Utilice convenientemente el modo de consumidor del productor Use múltiples colas + colas para admitir modos de consumo múltiples de un productor normal para admitir el modo de cola de consumo de prioridad para admitir el modo de trabajo de límite de flujo

Instalar

pip install thread_worker

Trabajar

Worker Trabajo múltiple ordinario
Trabajo de cola de consumo de prioridad PriorWorker
LimitWorker actual Límite trabajo

usar

Inicialización, los siguientes parámetros están disponibles

    def __init__ ( self , consumer_func , consumer_count , logger , block , timeout ):
        '''
        @params consumer_func consumer_func default None
        @params consumer_count Work quantity
        @params logger You can customize the logger. The default is logging
        @params block You can set whether the thread ends when the queue is empty 
                By default, the thread will end when the queue is empty and the thread task is completed
                You can set it to true to enable work to monitor tasks all the time
        @params timeout The timeout for obtaining data in the queue is 5S by default, that is, if the data is not obtained within 5S, it will end

        '''

Consumer_func La función del consumidor predeterminada a ninguno. Si se agrega, se considera que el consumidor se inicia primero y luego se genera la tarea. Consumer_count controla cuántos consumidores están allí.
Logger es establecer la salida del registro
El bloque es para controlar si es el modo de bloqueo. Si se establece en True, el hilo de trabajo continuará ejecutándose hasta que finalice el proceso principal. De lo contrario, cuando la cola está vacía y las tareas del hilo se completan, el trabajo finalizará.
El tiempo de espera es el tiempo de espera para obtener datos de la cola. El tiempo de espera predeterminado es 5s. Si la unión es verdadera, el trabajo llevará la unión a False nuevamente y finalizará el trabajo actual.

Hay dos escenarios de uso para el modelo de productor y consumidor.

Una es ejecutar el consumidor primero y luego generarlo. Puede generarlo y consumirlo continuamente. Es adecuado para tareas de escaneo de agentes pasivos o un proceso principal controla el global y luego genera tareas.

Una es generar todas las tareas a la vez y entregarlas al consumidor para ejecutar la tarea que es adecuada para generar pruebas a la vez y luego ejecutarla.

Bloquear tareas múltiples a través del método Work.IS_END

1. Definir las funciones del consumidor
2. Ejemplo un Work
3. A la función del consumidor a través Work put datos de la instancia
4. Bloquear el programa a través del método Work.is_end()

Trabajador básico

 from thread_worker import Worker
import requests
import time
def consumer ( iid ):
    url = "https://www.baidu.com/{0}" . format ( iid )
    resp = requests . get ( url = url )
    print ( resp . request . url )
# 不需要阻塞
w = Worker ( consumer , consumer_count = 1 , block = False )
for iid in range ( 10 ):
    w . put ( iid )
# 这里通过 is_end 方法来阻塞程序 
while not w . is_end ():
    time . sleep ( 3 )

Crear un trabajo que limite el flujo

Al igual que el trabajo predeterminado, hay un parámetro limit_time adicional al crearlo. Al establecer limit_time y consumer_count , controla cuántos trabajos ejecutar dentro de un tiempo de unidad.
El limit_time predeterminado_time es 1s

 from thread_worker import LimitWorker
import requests
import time
def consumer ( iid ):
    url = "https://www.baidu.com/{0}" . format ( iid )
    resp = requests . get ( url = url )
    print ( resp . request . url )
# limit_time 是limit_time 秒内有 consumer_count个消费者
w = LimitWorker ( consumer , consumer_count = 1 , block = False , limit_time = 3 )
for iid in range ( 10 ):
    w . put ( iid )
# 这里通过 is_end 方法来阻塞程序 
while not w . is_end ():
    time . sleep ( 3 )

La salida es así. Solo se enviará una solicitud cada 3 segundos, es decir, la frecuencia se controlará.

 send: 2022-04-08 10:13:04 904000
https://www.baidu.com/0
send: 2022-04-08 10:13:07 904000
https://www.baidu.com/1
send: 2022-04-08 10:13:10 904000
https://www.baidu.com/2
send: 2022-04-08 10:13:13 904000
https://www.baidu.com/3
send: 2022-04-08 10:13:16 904000
https://www.baidu.com/4
send: 2022-04-08 10:13:19 904000
https://www.baidu.com/5
send: 2022-04-08 10:13:22 904000
https://www.baidu.com/6
send: 2022-04-08 10:13:25 904000
https://www.baidu.com/7
send: 2022-04-08 10:13:28 904000
https://www.baidu.com/8
send: 2022-04-08 10:13:31 904000
https://www.baidu.com/9

Primero ejecute el consumidor y escuche la cola y luego genere continuamente tareas

En este escenario, los agentes pasivos o algunos procesos principales se utilizarán en programas que se ejecutan durante mucho tiempo. La diferencia es que el método Work.is_end () no es necesario para bloquear el bloque o establecer el bloque. El valor predeterminado es verdadero.

 from thread_worker import Worker
import requests
import time
def consumer ( iid ):
    url = "https://www.baidu.com/{0}" . format ( iid )
    resp = requests . get ( url = url )
    print ( resp . request . url )
# block默认就是True的
w = Worker ( consumer , consumer_count = 1 )
for iid in range ( 10 ):
    w . put ( iid )
# 手动阻塞
while True :
    pass