Obtenir tous les articles de la file d'attente

Question

J'ai un fil qui écrit les résultats dans une file d'attente.

Dans un autre thread (GUI), je vérifie périodiquement (dans l'événement IDLE) s'il y a des résultats dans la file d'attente, comme ceci:

def queue_get_all(q):
    items = []
    while 1:
        try:
            items.append(q.get_nowait())
        except Empty, e:
            break
    return items

Est-ce une bonne façon de le faire?

Modifier:

  

Je demande parce que parfois le   le fil d'attente se coince pendant quelques temps   secondes sans sortir de nouveau   résultats.

Le " bloqué " Le problème s’est avéré dû au fait que j’effectuais le traitement dans le gestionnaire d’événements inactif, sans vérifier que de tels événements sont réellement générés en appelant wx.WakeUpIdle , comme recommandé.

Answer 1

Je serais très surpris que l'appel get_nowait () provoque la pause en ne renvoyant pas si la liste était vide.

Est-il possible que vous publiiez un grand nombre d'éléments (peut-être gros?) entre les contrôles, ce qui signifie que le thread récepteur a beaucoup de données à extraire de la file d'attente ? Vous pouvez essayer de limiter le nombre que vous récupérez en un seul lot:

def queue_get_all(q):
    items = []
    maxItemsToRetreive = 10
    for numOfItemsRetrieved in range(0, maxItemsToRetreive):
        try:
            if numOfItemsRetrieved == maxItemsToRetreive:
                break
            items.append(q.get_nowait())
        except Empty, e:
            break
    return items

Cela limiterait le fil de réception à tirer jusqu'à 10 éléments à la fois.

Answer 2

Si vous retirez toujours tous les éléments disponibles de la file d'attente, l'utilisation d'une file d'attente présente-t-elle un intérêt réel plutôt que simplement une liste avec un verrou? c'est-à-dire:

from __future__ import with_statement
import threading

class ItemStore(object):
    def __init__(self):
        self.lock = threading.Lock()
        self.items = []

    def add(self, item):
        with self.lock:
            self.items.append(item)

    def getAll(self):
        with self.lock:
            items, self.items = self.items, []
        return items

Si vous les tirez également individuellement et utilisez le comportement de blocage pour les files d'attente vides, vous devez utiliser Queue, mais votre cas d'utilisation semble beaucoup plus simple et pourrait être mieux servi par l'approche ci-dessus.

[Edit2] Je n'avais pas compris le fait que vous interrogiez la file d'attente à partir d'une boucle inactive et que, dans votre mise à jour, je vois que le problème n'est pas lié à un conflit. L'approche ci-dessous n'est pas vraiment pertinente pour votre problème. Je l'ai laissé au cas où quelqu'un trouverait une variante bloquante de cet utile:

Dans les cas où vous souhaitez bloquer jusqu'à ce que vous obteniez au moins un résultat, vous pouvez modifier le code ci-dessus afin d'attendre que les données deviennent disponibles en étant signalées par le thread producteur. P. ex.

class ItemStore(object):
    def __init__(self):
        self.cond = threading.Condition()
        self.items = []

    def add(self, item):
        with self.cond:
            self.items.append(item)
            self.cond.notify() # Wake 1 thread waiting on cond (if any)

    def getAll(self, blocking=False):
        with self.cond:
            # If blocking is true, always return at least 1 item
            while blocking and len(self.items) == 0:
                self.cond.wait()
            items, self.items = self.items, []
        return items

Answer 3

Je pense que le moyen le plus simple de sortir tous les éléments de la file d'attente est le suivant:

def get_all_queue_result(queue):

    result_list = []
    while not queue.empty():
        result_list.append(queue.get())

    return result_list

Answer 4

Je vois que vous utilisez get_nowait () qui, selon la documentation, "renvoie [s] un élément s'il est immédiatement disponible, sinon déclenche une exception vide"

.

Maintenant, il vous arrive de sortir de la boucle lorsqu'une exception vide est levée. Ainsi, si aucun résultat n'est immédiatement disponible dans la file d'attente, votre fonction renvoie une liste d'éléments vides.

Y a-t-il une raison pour laquelle vous n'utilisez pas la méthode get () à la place? La méthode get_nowait () peut échouer car la file d'attente traite une demande put () au même moment.

Answer 5

Si vous avez fini d'écrire dans la file d'attente, qsize devrait faire l'affaire sans avoir à vérifier la file d'attente à chaque itération.

responseList = []
for items in range(0, q.qsize()):
    responseList.append(q.get_nowait())

Answer 6

La méthode la plus simple consiste à utiliser une compréhension de liste:

items = [q.get() for _ in range(q.qsize())]

L'utilisation de la fonction range est généralement mal vue, mais je n'ai pas encore trouvé de méthode plus simple.