Quelles sont les manières d'utiliser les files d'attente Python ?

Il existe environ trois modules de file d'attente utilisés en python :

1 from queue import Queue

Ce module convient à la communication inter-thread, mais ne peut pas être utilisé. pour les processus de communication entre.

Exemple de code 1 : [Remarque : il y a une erreur dans le code en ce moment ! ! ! 】

import time
import threading
from queue import Queue
def task_func():
    global queue
    while queue.qsize() > 0:
        x = queue.get()
        print(f"num: {x}")
        time.sleep(0.1)
def producer_data():
    global queue
    for i in range(100):
        queue.put(i)
        time.sleep(0.1)
if __name__ == '__main__':
    queue = Queue()
    producer_thread = threading.Thread(target=producer_data)
    producer_thread.start()
    thread_list = []
    for i in range(5):
        thread = threading.Thread(target=task_func)
        thread.start()
        thread_list.append(thread)
    for thread in thread_list:
        thread.join()
    print("主程序执行结束！")

Remarque : L'écriture ci-dessus :

    while queue.qsize() > 0:
        x = queue.get()

Lorsque la vitesse du producteur n'est pas aussi rapide que la vitesse du consommateur, le code du consommateur ci-dessus se terminera tôt, ce qui empêchera le producteur de consommer rapidement.

    while True:
        x = queue.get()

Il y a aussi un problème avec cette façon d'écrire. À ce stade, la file d'attente du consommateur surveillera toujours si la file d'attente du producteur contient des données, ce qui entraîne le blocage permanent du thread et du programme. sera bloqué et ne s'arrêtera pas, ce qui constitue un grave gaspillage de ressources système. Si vous utilisez une bibliothèque de tâches planifiées telle qu'apscheduler, la tâche planifiée ne sera pas démarrée.

En fait, le paramètre timeout est fourni dans la méthode put() ou get() dans la file d'attente. L'utilisation de ce paramètre peut résoudre efficacement les problèmes d'incapacité de consommation et de blocage de thread mentionnés ci-dessus.

Exemple de code 2 :

import time
import threading
from queue import Queue
def task_func():
    global queue
    while True:
        x = queue.get(timeout=10)
        print(f"num: {x}")
def producer_data():
    global queue
    for i in range(100):
        queue.put(i)
        time.sleep(0.1)
if __name__ == '__main__':
    queue = Queue()
    producer_thread = threading.Thread(target=producer_data)
    producer_thread.start()
    thread_list = []
    for i in range(5):
        thread = threading.Thread(target=task_func)
        thread.start()
        thread_list.append(thread)
    for thread in thread_list:
        thread.join()
    print("主程序执行结束！")

Résultat du running :

Selon différentes situations, vous peut définir la valeur du délai d'attente en fonction de la situation réelle. Si vous utilisez une tâche planifiée, vous pouvez utiliser le délai d'attente et le programme n'arrêtera pas de lever une exception.

2. à partir de la file d'attente d'importation multitraitement

Ce module est utilisé pour les processus, mais ne peut pas être utilisé pour les pools de processus

Exemple de code : #🎜🎜 #

import time
from multiprocessing import Process, Queue
import queue
def producer(queue):
    queue.put("a")
    time.sleep(2)
def consumer(queue):
    time.sleep(2)
    data = queue.get()
    print(data)
if __name__ == "__main__":
    # queue = queue.Queue()
    queue = Queue()
    my_producer = Process(target=producer, args=(queue, ))
    my_consumer = Process(target=consumer, args=(queue, ))
    my_producer.start()
    my_consumer.start()
    my_producer.join()
    my_consumer.join()
# 使用queue模块的Queue()会报错
# 使用multiprocessing中的Queue(),正确输出a

Résultat d'exécution :

3.

import time
from multiprocessing import Process, Queue, Pool, Manager
def producer(queue):
    queue.put("a")
    time.sleep(2)
def consumer(queue):
    time.sleep(2)
    data = queue.get()
    print(data)
if __name__ == "__main__":
    # queue = Queue()
    queue = Manager().Queue()
    pool = Pool()
    # pool中的进程间通信需要使用Manager
    pool.apply_async(producer, args=(queue, ))
    pool.apply_async(consumer, args=(queue, ))
    pool.close()
    pool.join()

Résultat de l'exécution :

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!