Procesamiento asíncrono

1Introducción

Temma ofrece Asynk, una forma sencilla de realizar procesamiento asíncrono. Este sistema se basa en el uso del componente de inyección de dependencias, extendiéndolo para que las llamadas a los objetos gestionados por el componente se procesen de forma asíncrona.

Ejemplo:

// llamada síncrona vía loader
$this->_loader->MyObject->myMethod($param1, $param2);

// llamada asíncrona idéntica
$this->_loader->asynk->MyObject->myMethod($param1, $param2);

Al igual que con el componente de inyección de dependencias, es posible usar la notación de array para ejecutar un objeto ubicado en un namespace: 

// llamada asíncrona
$this->_loader->asynk['\My\Name\Space\MyObject']->myMethod($param1, $param2);

Las tareas pueden ejecutarse de varias formas distintas:

  • mediante workers de Temma (programas que se ejecutan en segundo plano)
  • mediante un script de Temma ejecutado por crontab cada minuto
  • mediante un script de Temma ejecutado por xinetd cada vez que se lanza una tarea

Para la ejecución por worker o crontab, las tareas pueden obtenerse de una cola de mensajes (Beanstalkd o AWS SQS) o de una base de datos (MySQL o Redis). Para la ejecución por xinetd, solo es posible el almacenamiento en base de datos (MySQL o Redis).

Ten en cuenta los límites de los datos almacenados para cada tarea:

  • 16 MB para MySQL
  • 512 MB para Redis
  • 64 KB para Beanstalk
  • 256 KB para SQS

Beanstalkd es la cola de mensajes recomendada para ejecutar tareas con Asynk.
También puedes usar una cola Amazon SQS, pero esto exige que los workers se conecten regularmente para ver si hay tareas pendientes de ejecución, y cada petición realizada tiene un costo.

La solución más sencilla es almacenar las tareas en una base de datos, y ejecutarlas mediante crontab. Para una ejecución rápida de las tareas, lo más cercano posible al tiempo real, recomendamos añadir la ejecución mediante servidor xinetd.

2Configuración

2.1Principio

La configuración se basa en una configuración extendida x-asynk en el archivo etc/temma.php, que contiene dos parámetros:

  • transport: el nombre de la fuente de datos usada para transmitir las tareas.
    El comportamiento variará según el tipo de fuente de datos:
    • no definida: procesamiento por crontab o worker con almacenamiento en MySQL o Redis
    • socket: procesado por xinetd con almacenamiento en MySQL o Redis
    • Beanstalk: procesado por la cola de mensajes Beanstalkd
    • SQS: procesado por la cola de mensajes AWS SQS
  • storage: el nombre de la fuente de datos que almacenará los mensajes hasta que sean procesados, en caso de que no haya cola de mensajes:
    • MySQL: almacenamiento en una base de datos MySQL.
    • Redis: almacenamiento en una base de datos Redis.

Según el valor, cualquiera de los dos parámetros puede ser opcional.

Estas son las combinaciones posibles:

transport storage Procesamiento
MySQL Procesamiento con crontab o workers
Redis
socket MySQL Procesado por xinetd (+ crontab opcional)
socket Redis
Beanstalk Procesamiento mediante workers
SQS

Asynk escribe mensajes de log usando la clase de log Temma/Asynk.

Aquí tienes un ejemplo de archivo de configuración: 

<?php

return [
    'application' => [
        // fuentes de datos
        'dataSources' => [
            // cola de mensajes Amazon SQS
            'sqs' => 'sqs://AKXYZ:PWD@sqs.eu-west-3.amazonaws.com/123456789012/queue_name',
        ]
    ],
    // umbrales de los mensajes de log
    'loglevels' => [
        'Temma/Base'  => 'ERROR',
        'Temma/Web'   => 'WARN',
        'Temma/Asynk' => 'NOTE',
    ],
    // configuración de Asynk
    'x-asynk' => [
        // transporte: cola SQS
        'transport' => 'sqs',
    ],
];

2.2Almacenamiento en MySQL

Si eliges almacenar tus tareas en una base de datos MySQL, tendrás que crear la tabla que contiene las tareas.

Aquí tienes la consulta para crear esta tabla:

CREATE TABLE Task (
    id             INT UNSIGNED NOT NULL AUTO_INCREMENT,
    dateCreation   DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
    dateUpdate     DATETIME ON UPDATE CURRENT_TIMESTAMP,
    status         ENUM('waiting', 'reserved', 'processing', 'error') NOT NULL DEFAULT 'waiting',
    token          CHAR(16) CHARACTER SET ascii COLLATE ascii_general_ci,
    target         TINYTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
    action         TINYTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
    data           MEDIUMTEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci,
    PRIMARY KEY (id),
    INDEX status (status),
    INDEX token (token)
);

La clave primaria (campo id) puede ser de tipo BIGINT si necesitas gestionar más de 4 mil millones de tareas.

Si la tabla se almacena en una base de datos distinta de la de la conexión, o si se llama de otra forma que no sea Task, o si los campos tienen otros nombres, esto puede indicarse en la configuración extendida x-asynk en etc/temma.php. 

<?php

return [
    'x-asynk' => [
        'base'   => 'asynk_db',
        'table'  => 'asynk_tasks',
        'id'     => 'task_id',
        'status' => 'task_status',
        'token'  => 'task_token',
        'target' => 'task_target',
        'action' => 'task_action',
        'data'   => 'task_data',
    ]
];
  • Línea 5: Nombre de la base de datos que contiene la tabla.
  • Línea 6: Nombre de la tabla.
  • Línea 7: Nombre del campo que contiene la clave primaria.
  • Línea 8: Nombre del campo que contiene el estado.
  • Línea 9: Nombre del campo que contiene el token de reserva.
  • Línea 10: Nombre del campo que contiene el objeto de destino.
  • Línea 11: Nombre del campo que contiene el método de destino.
  • Línea 12: Nombre del campo que contiene los parámetros serializados.

También es posible usar un DAO personalizado: 

<?php

return [
    'x-asynk' => [
        'dao' => '\MyApp\AsynkDao'
    ]
];

3Procesamiento por crontab

3.1Introducción a crontab

El procesamiento por crontab es muy sencillo de implementar. El demonio crontab está instalado o es instalable en todos los sistemas Unix, y es independiente de otro software como una cola de mensajes.

El procesamiento por crontab se ejecuta cada minuto. Esto puede causar un ligero retraso en el procesamiento de las tareas. Si necesitas un procesamiento sin latencia, puedes añadir el procesamiento por xinetd (ver más abajo).

3.2Configuración de Temma para crontab

El almacenamiento usado para guardar las tareas debe declararse en la configuración extendida x-asynk.

Ejemplo de archivo etc/temma.php, con almacenamiento en Redis:

<?php

return [
    'application' => [
        'dataSources' => [
            'ndb' => 'redis://localhost'
        ]
    ],
    'x-asynk' => [
        'storage'   => 'ndb'
    ],
];

3.3Configuración de crontab

Hay dos formas de configurar el crontab:

  • Modifica el contenido del archivo etc/asynk/crontab para adaptar la ruta a la raíz de tu proyecto.
    Después copia este archivo a /etc/cron.d/ (con un nombre adecuado), por ejemplo con el comando:
    sudo cp /path/to/project/etc/asynk/crontab /etc/cron.d/my_project
  • O añade la siguiente línea (adaptando la ruta) al crontab del usuario deseado:
    * * * * *    cd /path/to/project/; bin/comma 'Asynk\Worker/crontab'

4Procesamiento por xinetd

4.1Introducción a xinetd

Xinetd es un “superdemonio” que escucha en varios puertos de red. Cada vez que recibe una conexión entrante, lanza el programa asociado, y se encarga de los intercambios de red.
Para la gestión de Asynk, xinetd escucha por defecto en el puerto 11137.

Cuando Asynk usa xinetd, las tareas asíncronas se procesan de inmediato. Sin embargo, hay que tener en cuenta dos cosas:

  • Si necesitas gestionar un número muy elevado de tareas simultáneas, xinetd mostrará su límite. En este caso, recomendamos usar una cola de mensajes (Beanstalkd o SQS).
  • Puede ocurrir que xinetd no sea capaz de gestionar ciertas tareas (el demonio no está en ejecución o está saturado). Por ello es recomendable combinar el procesamiento por xinetd con el procesamiento por crontab (ver más arriba), de modo que las tareas no procesadas por xinetd sean procesadas por el crontab.

4.2Configuración de Temma para xinetd

En la configuración, se requiere una fuente de datos socket, que Asynk usará para conectarse a xinetd. El servidor xinetd puede estar en el servidor local o en una máquina remota. Por defecto, el puerto de conexión usado por Asynk es 11137.

Ejemplo de archivo etc/temma.php, con almacenamiento en MySQL:

<?php

return [
    'application' => [
        'dataSources' => [
            'sock' => 'tcp://localhost:11137',
            'db'   => 'mysql://user:password@localhost'
        ]
    ],
    'x-asynk' => [
        'transport' => 'sock',
        'storage'   => 'db'
    ],
];

4.3Configuración de xinetd

Para configurar xinetd, edita el archivo etc/asynk/xinetd para adaptar la ruta, luego cópialo al directorio /etc/xinetd.d/ (con un nombre adecuado), por ejemplo con el comando:

sudo cp /path/to/project/etc/asynk/xinetd /etc/xinetd.d/my_project

4.4(opcional) Configuración de crontab

Usar el crontab además de xinetd garantiza que todas las tareas se procesen, incluso si xinetd no está en ejecución o está saturado.

Hay dos formas de configurar el crontab:

  • Modifica el contenido del archivo etc/asynk/crontab para adaptar la ruta a la raíz de tu proyecto.
    Después copia este archivo a /etc/cron.d/ (con un nombre adecuado), por ejemplo con el comando:
    sudo cp /path/to/project/etc/asynk/crontab /etc/cron.d/my_project
  • O añade la siguiente línea (adaptando la ruta) al crontab del usuario deseado:
    * * * * *    cd /path/to/project/; bin/comma 'Asynk\Worker/crontab'

5Procesamiento por worker

5.1Presentación del worker

Los workers son programas que se ejecutan en segundo plano. Puedes ejecutar tantos workers como quieras. Si solo se ejecuta un worker, puede considerarse un demonio de procesamiento.

Los workers se conectan a la fuente de datos (cola de mensajes o base de datos) para obtener las tareas que deben ejecutarse. Las obtienen una a una, y las procesan secuencialmente. Si tienes un gran número de tareas que procesar, lo mejor es tener varios workers ejecutándose en paralelo, ya que de lo contrario las tareas pueden acumularse más rápido de lo que se pueden procesar.

Es importante asegurarse de que un número mínimo de workers se ejecute en todo momento, para evitar el riesgo de que las tareas no se procesen. Es posible usar un supervisor como Supervisord, que reiniciará automáticamente los workers si no se garantiza el número mínimo de instancias.

5.2Configuración de Temma para workers

La configuración de Temma debe contener información sobre el almacenamiento y, si procede, el transporte de las tareas.

Ejemplo de archivo etc/temma.php, con almacenamiento en MySQL:

<?php

return [
    'application' => [
        'dataSources' => [
            'db'        => 'mysql://user:password@localhost'
        ]
    ],
    'x-asynk' => [
        'storage'   => 'db'
    ],
];

Otro ejemplo de archivo etc/temma.php, con transporte Beanstalkd:

<?php

return [
    'application' => [
        'dataSources' => [
            'beanstalk' => 'beanstalk://localhost/tube3'
        ]
    ],
    'x-asynk' => [
        'transport' => 'beanstalk'
    ],
];

Otro ejemplo de archivo etc/temma.php, con transporte Amazon SQS:

<?php

return [
    'application' => [
        'dataSources' => [
            'sqs' => 'sqs://AKXYZ:PWD@sqs.eu-west-3.amazonaws.com/123456789012/queue_name'
        ]
    ],
    'x-asynk' => [
        'transport' => 'sqs'
    ],
];

Por defecto, los workers en modo de sondeo esperan 60 segundos entre dos conexiones para recuperar las tareas pendientes. Esto se aplica a los workers que se conectan a una cola Amazon SQS o a una base de datos MySQL o Redis; no se aplica a las colas Beanstalkd.

Este retraso puede modificarse usando el parámetro loopDelay en la configuración extendida x-asynk: 

<?php

return [
    'application' => [
        'dataSources' => [
            'db'        => 'mysql://user:password@localhost'
        ]
    ],
    'x-asynk' => [
        'storage'   => 'db',
        // retraso de 90 segundos entre dos comprobaciones
        'loopDelay' => 90
    ],
];

5.3Configuración de Supervisor

El uso de Supervisor es opcional, pero puede ser útil para garantizar que los workers estén en ejecución, y que se reinicien si ocurre un problema.

Copia el archivo etc/asynk/supervisor.conf a /etc/supervisor/conf.d/asynk.conf con el siguiente comando: 

sudo cp /path/to/project/etc/asynk/supervisor.conf /etc/supervisor/conf.d/asynk.conf

Luego modifica los siguientes parámetros en el archivo:

  • command: Indica la ruta correcta a la raíz de tu proyecto.
    Ejemplo: command=/path/to/project/bin/comma 'Asynk\Worker'
  • numprocs: Define el número de workers que se ejecutarán al mismo tiempo.
    Ejemplo: numprocs=5

Luego fuerza a Supervisor a tener en cuenta esta configuración:

sudo supervisorctl reread
sudo supervisorctl update