Warten auf eine Einzel-RabbitMQ-Meldung mit einem timeout

Ich würde gerne eine Nachricht senden, die eine RabbitMQ-server und warten dann auf eine Antwort-Nachricht ("reply-to" - Warteschlange). Natürlich, ich will nicht ewig warten, falls die Anwendung die Verarbeitung dieser Nachrichten ist down - muss es ein timeout. Es klingt wie eine sehr einfache Aufgabe sein, doch ich kann nicht einen Weg finden, dies zu tun. Ich habe jetzt dieses problem mit beiden py-amqplib und die RabbitMQ .NET-client.

Die beste Lösung, die ich habe, so weit ist die Umfrage mit basic_get mit sleep in-zwischen, aber das ist ziemlich hässlich:

def _wait_for_message_with_timeout(channel, queue_name, timeout):
    slept = 0
    sleep_interval = 0.1

    while slept < timeout:
        reply = channel.basic_get(queue_name)
        if reply is not None:
            return reply

        time.sleep(sleep_interval)
        slept += sleep_interval

    raise Exception('Timeout (%g seconds) expired while waiting for an MQ response.' % timeout)

Sicherlich gibt es einige bessere Weg?

InformationsquelleAutor EMP | 2010-05-10

  1. 8

    Habe ich nur noch timeout support für amqplib im carrot.

    Dies ist eine Unterklasse von amqplib.client0_8.Connection:

    http://github.com/ask/carrot/blob/master/carrot/backends/pyamqplib.py#L19-97

    wait_multi ist eine version von channel.wait erhalten können auf eine beliebige Anzahl
    der Kanäle.

    Ich denke, das könnte zusammengefasst werden upstream-irgendwann.

    Nun das ist nenne ich eine "Super Antwort": "es ist Fix"! Annahme - in der Hoffnung, dass es wird zusammengeführt amqplib.
    haha 🙂 lustig 🙂

    InformationsquelleAutor asksol

  2. 8

    Hier ist, was ich am Ende machen .NET-client:

    protected byte[] WaitForMessageWithTimeout(string queueName, int timeoutMs)
{
    var consumer = new QueueingBasicConsumer(Channel);
    var tag = Channel.BasicConsume(queueName, true, null, consumer);
    try
    {
        object result;
        if (!consumer.Queue.Dequeue(timeoutMs, out result))
            throw new ApplicationException(string.Format("Timeout ({0} seconds) expired while waiting for an MQ response.", timeoutMs / 1000.0));

        return ((BasicDeliverEventArgs)result).Body;
    }
    finally
    {
        Channel.BasicCancel(tag);
    }
}

    Leider kann ich nicht das gleiche tun mit py-amqplib, weil seine basic_consume Methode nicht, rufen Sie die callback-es sei denn, Sie rufen channel.wait() und channel.wait() unterstützt keine timeouts! Diese dumme Beschränkung (die ich am laufen zu halten) bedeutet, dass, wenn Sie nicht erhalten eine weitere Nachricht dein thread ist eingefroren, für immer.

    InformationsquelleAutor EMP

  3. 2

    Gibt es ein Beispiel hier mit qpid mit einem msg = q.get(timeout=1) sollten tun, was Sie wollen. Sorry, ich weiß nicht, was andere AMQP-client-Bibliotheken implementieren timeouts (und insbesondere ich weiß nicht, die zwei bestimmte diejenigen, die Sie erwähnt).

    Suche an der Quelle des qpid es scheint, verwenden die exakt gleichen Ansatz wie die .NET-client: basic_consume mit einer Warteschlange und warten auf die queue mit einem timeout. Sieht aus wie das ist, was ich werde zu tun haben.

    InformationsquelleAutor Alex Martelli

  4. 1

    Diese scheint zu brechen, die die ganze Idee der asynchronen Verarbeitung, aber wenn man muss, glaube ich, der richtige Weg, es zu tun, ist die Verwendung eines RpcClient.

    Während RpcClient an sich ist nicht nützlich für mich, Blick auf die Umsetzung zeigt sich der Ansatz zu verwenden: erstellen Sie eine QueueingBasicConsumer und warten auf die Warteschlange, unterstützt ein timeout. Dies ist nicht so Komplex in .NET wie ich es befürchtet habe.

    InformationsquelleAutor duffymo

  5. 1

    Kaninchen nun ermöglicht Ihnen das hinzufügen timeout-Ereignisse. Einfach wickeln Sie Ihren code in einen try-catch und dann auslösen von Ausnahmen in den TimeOut und Disconnect-Handler:

    try{
    using (IModel channel = rabbitConnection.connection.CreateModel())
    {
        client = new SimpleRpcClient(channel, "", "", queue);
        client.TimeoutMilliseconds = 5000; // 5 sec. defaults to infinity
        client.TimedOut += RpcTimedOutHandler;
        client.Disconnected += RpcDisconnectedHandler;
        byte[] replyMessageBytes = client.Call(message);
        return replyMessageBytes;
    }
}
catch (Exception){
    //Handle timeout and disconnect here
}
private void RpcDisconnectedHandler(object sender, EventArgs e)
{
     throw new Exception("RPC disconnect exception occured.");
}

private void RpcTimedOutHandler(object sender, EventArgs e)
{
     throw new Exception("RPC timeout exception occured.");
}

    InformationsquelleAutor Deadly

Schreibe einen Kommentar