job - azure functions tutorial

Sondeo de una cola de bus de servicio de Azure desde Azure WebJob utilizando Node.js

Question

Intentando sondear una cola de bus de servicio de Azure usando un WebJob escrito en Node.js. Creé 2 WebJobs. El primero es bajo demanda y envía 10 mensajes únicos a la cola. El segundo trabajo es continuo y consulta la lista de mensajes.

Encontrando los siguientes problemas:

1. El sondeo es LENTO. Toma un promedio de aproximadamente 10 minutos para recibir 10 mensajes. Vea los detalles del registro de muestra a continuación. Básicamente inutilizable a esta velocidad. Todo el retraso consiste en obtener una respuesta de receiveQueueMessage . Los tiempos de respuesta varían de 0 segundos a ~ 120 segundos, con un promedio de 60 segundos.

2. Los mensajes se están recibiendo en un orden aleatorio. No es FIFO.

3. A veces los mensajes se reciben dos veces, aunque se están leyendo en modo ReceiveAndDelete (he intentado sin parámetro de modo de lectura que debería ser ReceiveAndDelete, con {isReceiveAndDelete:true} y con {isPeekLock:false} con los mismos resultados).

4. Cuando la cola está vacía, debe mantener la solicitud de recepción abierta durante un día, pero siempre regresa con un mensaje de error sin mensaje después de 230 segundos. De acuerdo con la documentación, el máximo es de 24 días, así que no sé de dónde vienen 230 segundos:

El tiempo de espera máximo para una operación de recepción de bloqueo en las colas del bus de servicio es de 24 días. Sin embargo, los tiempos de espera basados ​​en REST tienen un valor máximo de 55 segundos.

Básicamente, nada funciona como se anuncia. ¿Qué estoy haciendo mal?

Enviar mensaje Prueba de trabajo:

var uuid = require(''node-uuid'');
var azure = require(''azure'');
var serviceBus = azure.createServiceBusService(process.env.busSearchConnectionString);
var messagesToSend = 10;

sendMessage(0);

function sendMessage(count)
{
    var message = {
        body: ''test message'',
        customProperties: {
            message_number: count,
            sent_date: new Date
        },
        brokerProperties: {
            MessageId: uuid.v4() //ensure that service bus doesn''t think this is a duplicate message
        }
    };

    serviceBus.sendQueueMessage(process.env.busSearchQueueName, message, function(err) {

        if (!err) {
            console.log(''sent test message number '' + count.toString());
        } else {
            console.error(''error sending message: '' + err);
        }

    });

    //wait 5 seconds to ensure messages are received by service bus in correct order
    if (count < messagesToSend) {
        setTimeout(function(newCount) {
            //send next message
            sendMessage(newCount);
        }, 5000, count+1);
    }
}    

Recibir mensaje de trabajo continuo:

console.log(''listener job started'');
var azure = require(''azure'');
var serviceBus = azure.createServiceBusService(process.env.busSearchConnectionString);
listenForMessages(serviceBus);

function listenForMessages(serviceBus)
{
    var start = process.hrtime();
    var timeOut = 60*60*24; //long poll for 1 day
    serviceBus.receiveQueueMessage(process.env.busSearchQueueName, {timeoutIntervalInS: timeOut, isReceiveAndDelete: true}, function(err, message) {

        var end = process.hrtime(start);
        console.log(''received a response in %ds seconds'', end[0]);

        if (err) {

            console.log(''error requesting message: '' + err);
            listenForMessages(serviceBus);

        } else {

            if (message !== null && typeof message === ''object'' && ''customProperties'' in message && ''message_number'' in message.customProperties) {

                console.log(''received test message number '' + message.customProperties.message_number.toString());
                listenForMessages(serviceBus);

            } else {

                console.log(''invalid message received'');
                listenForMessages(serviceBus);

            }

        }

    });
}

Muestra de salida de registro:

[05/06/2015 21:50:14 > 8c2504: SYS INFO] Status changed to Running
[05/06/2015 21:50:14 > 8c2504: INFO] listener job started
[05/06/2015 21:51:23 > 8c2504: INFO] received a response in 1s seconds
[05/06/2015 21:51:23 > 8c2504: INFO] received test message number 0
[05/06/2015 21:51:25 > 8c2504: INFO] received a response in 2s seconds
[05/06/2015 21:51:26 > 8c2504: INFO] received test message number 4
[05/06/2015 21:51:27 > 8c2504: INFO] received a response in 1s seconds
[05/06/2015 21:51:27 > 8c2504: INFO] received test message number 7
[05/06/2015 21:51:28 > 8c2504: INFO] received a response in 0s seconds
[05/06/2015 21:51:29 > 8c2504: INFO] received test message number 9
[05/06/2015 21:51:49 > 8c2504: INFO] received a response in 20s seconds
[05/06/2015 21:51:49 > 8c2504: INFO] received test message number 1
[05/06/2015 21:53:35 > 8c2504: INFO] received a response in 106s seconds
[05/06/2015 21:53:35 > 8c2504: INFO] received test message number 1
[05/06/2015 21:54:26 > 8c2504: INFO] received a response in 50s seconds
[05/06/2015 21:54:26 > 8c2504: INFO] received test message number 5
[05/06/2015 21:54:35 > 8c2504: INFO] received a response in 9s seconds
[05/06/2015 21:54:35 > 8c2504: INFO] received test message number 9
[05/06/2015 21:55:28 > 8c2504: INFO] received a response in 53s seconds
[05/06/2015 21:55:28 > 8c2504: INFO] received test message number 2
[05/06/2015 21:57:26 > 8c2504: INFO] received a response in 118s seconds
[05/06/2015 21:57:26 > 8c2504: INFO] received test message number 6
[05/06/2015 21:58:28 > 8c2504: INFO] received a response in 61s seconds
[05/06/2015 21:58:28 > 8c2504: INFO] received test message number 8
[05/06/2015 22:00:35 > 8c2504: INFO] received a response in 126s seconds
[05/06/2015 22:00:35 > 8c2504: INFO] received test message number 3
[05/06/2015 22:04:25 > 8c2504: INFO] received a response in 230s seconds
[05/06/2015 22:04:25 > 8c2504: INFO] error requesting message: No messages to receive
[05/06/2015 22:08:16 > 8c2504: INFO] received a response in 230s seconds    
[05/06/2015 22:04:25 > 8c2504: INFO] error requesting message: No messages to receive

Answer 1

Desactivar el indicador de particiones de la cola del bus de servicio me funcionó también.

Con la cola particionada, algunos mensajes tuvieron retrasos de más de 30 minutos. Un cliente web simple de DotNet podría descargar todos los mensajes sin demoras. Sin embargo, tan pronto como se suponía que nodejs descargaba mensajes, solo el primer mensaje se descargaría sin problemas, luego aparecieron los retrasos. Jugar con nodejs para cambiar las opciones del agente http keepalive y el tiempo de espera del socket no mejoró la situación.

Después de detener nodejs, tuve que esperar varios minutos antes de que el cliente de DotNet realmente comenzara a trabajar sin problemas. Esto fue reproducible varias veces. También encontré que el programa simple webclient de DotNet mostraba problemas similares, después de haber sido iniciado y detenido varias veces seguidas.

De todos modos, su publicación me mostró la solución: apague la bandera particionada :)

Answer 2

Intente usar el amqp para leer los mensajes de la cola particionada del bus de servicio azure y esto funcionará para un tema / cola particionado y ni siquiera tiene que sondear mucho.

const AMQPClient = require(''amqp10'').Client;
const Policy = require(''amqp10'').Policy;

const protocol = ''amqps'';
const keyName = ''RootManageSharedAccessKey'';
const sasKey = ''your_key_goes_here'';
const serviceBusHost = ''namespace.servicebus.windows.net'';
const uri = `${protocol}://${encodeURIComponent(keyName)}:${encodeURIComponent(sasKey)}@${serviceBusHost}`;
const queueName = ''partitionedQueueName'';
const client = new AMQPClient(Policy.ServiceBusQueue);
client.connect(uri)
.then(() => Promise.all([client.createReceiver(queueName)]))
.spread((receiver) => {
    console.log(''--------------------------------------------------------------------------'');
    receiver.on(''errorReceived'', (err) => {
        // check for errors
        console.log(err);
    });
    receiver.on(''message'', (message) => {
        console.log(''Received message'');
        console.log(message);
        console.log(''----------------------------------------------------------------------------'');
    });
})
.error((e) => {
    console.warn(''connection error: '', e);
});

https://www.npmjs.com/package/amqp10

Answer 3

Y el problema era que la cola que estaba utilizando estaba particionada (la opción predeterminada al crear una cola en el portal de Azure). Una vez que creé una nueva cola que no estaba particionada, todo funcionó como se esperaba sin el retraso (aparte del raro tiempo de espera de 230 segundos en un largo intento de sondeo). Así que, básicamente, la biblioteca node.js no funciona para colas particionadas. En absoluto. Perdió muchos días pensando en eso. Dejaré esto aquí para otros.