thread_worker

thread_worker

Utilisez commodément le mode de consommation de producteur Utilisez des files d'attente multi-threading + pour prendre en charge le mode de file d'attente de la consommation de la consommation de la consommation à un seul producteur normal pour prendre en charge le mode de travail limite du flux

Installer

pip install thread_worker

Travail

Travaux multipliés ordinaires Worker
Travail de file d'attente de consommation prioritaire PriorWorker
LimitWorker COURANT LIMIT TRAVAIL

utiliser

Initialisation, les paramètres suivants sont 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 Consumer Fonction en fait par défaut. S'il est ajouté, il est considéré que le consommateur est démarré en premier, puis la tâche est générée. Consumer_Count contrôle le nombre de consommateurs.
Logger consiste à définir la sortie du journal
Le bloc doit contrôler s'il s'agit du mode de blocage. S'il est défini sur true, le thread de travail continuera de s'exécuter jusqu'à la fin du processus principal. Sinon, lorsque la file d'attente est vide et que les tâches de fil sont terminées, les travaux se termineront.
Le délai d'attente est le délai d'obtention pour obtenir des données de la file d'attente. Le délai d'expiration par défaut est de 5s. Si la jointure est vraie, l'œuvre reprendra à nouveau la jointure de False et mettra fin aux travaux actuels.

Il existe deux scénarios d'utilisation pour le modèle producteur et consommateur

L'une consiste à exécuter d'abord le consommateur, puis à le générer. Il peut en continu et le consommer. Il convient aux tâches de balayage d'agent passives ou un processus principal contrôle le global, puis génère des tâches.

L'une consiste à générer toutes les tâches à la fois et à les remettre au consommateur pour exécuter la tâche qui convient à la génération de tests à la fois, puis à l'exécuter.

Bloquer les tâches multithreads via la méthode work.is_end

1. Définir les fonctions des consommateurs
2. Exemple de Work
3. À la fonction des consommateurs à travers Work put les données de l'instance
4. Bloquer le programme via Work.is_end() Méthode

Travailleur de base

 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 )

Créer une œuvre qui limite le flux

Comme le travail par défaut, il y a un paramètre limit_time supplémentaire lors de la création. En définissant limit_time et consumer_count , il contrôle le nombre de travaux à exécuter dans un unité de temps.
La limit_time par défaut est 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 sortie est comme ça. Seule une demande sera envoyée toutes les 3 secondes, c'est-à-dire que la fréquence sera contrôlée.

 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

Exécutez d'abord le consommateur et écoutez la file d'attente, puis générez continuellement des tâches

Dans ce scénario, des agents passifs ou certains processus principaux seront utilisés dans des programmes qui fonctionneront longtemps. La différence est que la méthode work.is_end () n'est pas nécessaire pour bloquer le bloc ou définir le bloc. La valeur par défaut est vraie.

 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