Processamento assíncrono

1Introdução

O Temma oferece o Asynk, uma forma simples de realizar processamento assíncrono. Esse sistema é baseado no uso do componente de injeção de dependências, estendendo-o para que as chamadas a objetos gerenciados pelo componente sejam processadas de forma assíncrona.

Exemplo:

// chamada síncrona via loader
$this->_loader->MyObject->myMethod($param1, $param2);

// chamada assíncrona idêntica
$this->_loader->asynk->MyObject->myMethod($param1, $param2);

Assim como no componente de injeção de dependências, é possível usar a notação de array para executar um objeto localizado em um namespace: 

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

As tarefas podem ser executadas de várias formas diferentes:

  • pelos workers do Temma (programas executados em segundo plano)
  • por um script do Temma executado pelo crontab a cada minuto
  • por um script do Temma executado pelo xinetd cada vez que uma tarefa é lançada

Para a execução via worker ou crontab, as tarefas podem ser obtidas de uma fila de mensagens (Beanstalkd ou AWS SQS) ou de um banco de dados (MySQL ou Redis). Para a execução via xinetd, apenas o armazenamento em banco de dados (MySQL ou Redis) é possível.

Cuidado com os limites relativos aos dados armazenados para cada tarefa:

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

O Beanstalkd é a fila de mensagens recomendada para executar tarefas com o Asynk.
Você também pode usar uma fila Amazon SQS, mas isso exige que os workers se conectem regularmente para verificar se há tarefas aguardando execução, e há uma cobrança para cada requisição feita.

A solução mais simples é armazenar as tarefas em um banco de dados, e executá-las via crontab. Para uma execução rápida das tarefas, o mais próximo possível do tempo real, recomendamos adicionar a execução por meio do servidor xinetd.

2Configuração

2.1Princípio

A configuração é baseada em uma configuração estendida x-asynk no arquivo etc/temma.php, contendo dois parâmetros:

  • transport: o nome da fonte de dados usada para transmitir as tarefas.
    O comportamento varia de acordo com o tipo de fonte de dados:
    • indefinido: processamento por crontab ou worker com armazenamento MySQL ou Redis
    • socket: processado pelo xinetd com armazenamento MySQL ou Redis
    • Beanstalk: processado pela fila de mensagens Beanstalkd
    • SQS: processado pela fila de mensagens AWS SQS
  • storage: o nome da fonte de dados que armazenará as mensagens até que sejam processadas, caso não haja fila de mensagens:
    • MySQL: armazenamento em um banco de dados MySQL.
    • Redis: armazenamento em um banco de dados Redis.

Dependendo do valor, qualquer um dos parâmetros pode ser opcional.

Aqui estão as combinações possíveis:

transport storage Processamento
MySQL Processamento com crontab ou workers
Redis
socket MySQL Processado pelo xinetd (+ crontab opcional)
socket Redis
Beanstalk Processamento via workers
SQS

O Asynk grava mensagens de log usando a classe de log Temma/Asynk.

Aqui está um exemplo de arquivo de configuração: 

<?php

return [
    'application' => [
        // fontes de dados
        'dataSources' => [
            // fila de mensagens Amazon SQS
            'sqs' => 'sqs://AKXYZ:PWD@sqs.eu-west-3.amazonaws.com/123456789012/queue_name',
        ]
    ],
    // limites de nível das mensagens de log
    'loglevels' => [
        'Temma/Base'  => 'ERROR',
        'Temma/Web'   => 'WARN',
        'Temma/Asynk' => 'NOTE',
    ],
    // configuração do Asynk
    'x-asynk' => [
        // transport: fila SQS
        'transport' => 'sqs',
    ],
];

2.2Armazenamento MySQL

Se você optar por armazenar suas tarefas em um banco de dados MySQL, será necessário criar a tabela que conterá as tarefas.

Aqui está a consulta para criar essa tabela:

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)
);

A chave primária (campo id) pode ser do tipo BIGINT se você precisar gerenciar mais de 4 bilhões de tarefas.

Se a tabela estiver armazenada em um banco de dados diferente do da conexão, ou se ela tiver um nome diferente de Task, ou se os campos tiverem outros nomes, isso pode ser especificado na configuração estendida x-asynk em 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_token',
        'data'   => 'task_data',
    ]
];
  • Linha 5: nome do banco de dados que contém a tabela.
  • Linha 6: nome da tabela.
  • Linha 7: nome do campo que contém a chave primária.
  • Linha 8: nome do campo que contém o status.
  • Linha 9: nome do campo que contém o token de reserva.
  • Linha 10: nome do campo que contém o objeto de destino.
  • Linha 11: nome do campo que contém o método de destino.
  • Linha 12: nome do campo que contém os parâmetros serializados.

Também é possível usar um DAO personalizado: 

<?php

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

3Processamento via crontab

3.1Introdução ao crontab

O processamento via crontab é muito simples de implementar. O daemon do crontab está instalado ou pode ser instalado em todos os sistemas Unix, e é independente de outros softwares, como uma fila de mensagens.

O processamento via crontab é executado a cada minuto. Isso pode causar um pequeno atraso no processamento das tarefas. Se você precisar de um processamento sem latência, pode adicionar o processamento via xinetd (veja abaixo).

3.2Configuração do Temma para o crontab

O storage usado para salvar as tarefas deve ser declarado na configuração estendida x-asynk.

Exemplo de arquivo etc/temma.php, com armazenamento Redis:

<?php

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

3.3Configuração do crontab

Existem duas formas de configurar o crontab:

  • Modifique o conteúdo do arquivo etc/asynk/crontab para adaptar o caminho até a raiz do seu projeto.
    Em seguida, copie esse arquivo para /etc/cron.d/ (com um nome adequado), por exemplo com o comando:
    sudo cp /path/to/project/etc/asynk/crontab /etc/cron.d/my_project
  • Ou adicione a seguinte linha (adaptando o caminho) ao crontab do usuário desejado:
    * * * * *    cd /path/to/project/; bin/comma '\Temma\Cli\Asynk\Worker' crontab

4Processamento via xinetd

4.1Introdução ao xinetd

O xinetd é um “super-daemon” que escuta em várias portas de rede. Cada vez que recebe uma conexão de entrada, ele lança o programa associado e assume a responsabilidade pelas trocas de rede.
Para o gerenciamento do Asynk, o xinetd escuta na porta 11137 por padrão.

Quando o Asynk usa o xinetd, as tarefas assíncronas são processadas imediatamente. No entanto, há dois pontos a serem considerados:

  • Se você precisar lidar com um número muito grande de tarefas simultâneas, o xinetd mostrará seus limites. Nesse caso, recomendamos o uso de uma fila de mensagens (Beanstalkd ou SQS).
  • Pode acontecer de o xinetd não conseguir lidar com determinadas tarefas (o daemon não está em execução ou está saturado). Por isso, é aconselhável combinar o processamento via xinetd com o processamento via crontab (veja acima), para que as tarefas não processadas pelo xinetd sejam processadas pelo crontab.

4.2Configuração do Temma para o xinetd

Na configuração, é necessária uma fonte de dados socket, que o Asynk usará para se conectar ao xinetd. O servidor xinetd pode estar no servidor local ou em uma máquina remota. Por padrão, a porta de conexão usada pelo Asynk é 11137.

Exemplo de arquivo etc/temma.php, com armazenamento MySQL:

<?php

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

4.3Configuração do xinetd

Para configurar o xinetd, edite o arquivo etc/asynk/xinetd para adaptar o caminho, depois copie-o para o diretório /etc/xinetd.d/ (com um nome adequado), por exemplo com o comando:

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

4.4(opcional) Configuração do crontab

Usar o crontab em conjunto com o xinetd garante que todas as tarefas sejam processadas, mesmo que o xinetd não esteja em execução ou esteja saturado.

Existem duas formas de configurar o crontab:

  • Modifique o conteúdo do arquivo etc/asynk/crontab para adaptar o caminho até a raiz do seu projeto.
    Em seguida, copie esse arquivo para /etc/cron.d/ (com um nome adequado), por exemplo com o comando:
    sudo cp /path/to/project/etc/asynk/crontab /etc/cron.d/my_project
  • Ou adicione a seguinte linha (adaptando o caminho) ao crontab do usuário desejado:
    * * * * *    cd /path/to/project/; bin/comma '\Temma\Cli\Asynk\Worker' crontab

5Processamento por worker

5.1Apresentação do worker

Workers são programas executados em segundo plano. Você pode executar quantos workers desejar. Se apenas um worker estiver em execução, ele pode ser considerado um daemon de processamento.

Os workers se conectam à fonte de dados (fila de mensagens ou banco de dados) para obter as tarefas a serem executadas. Eles obtêm as tarefas uma a uma e as processam sequencialmente. Se você tiver um grande número de tarefas para processar, o ideal é ter vários workers em execução em paralelo; caso contrário, as tarefas podem se acumular mais rápido do que conseguem ser processadas.

É importante garantir que um número mínimo de workers esteja sempre em execução, para evitar o risco de tarefas não serem processadas. É possível usar um supervisor como o Supervisord, que reinicia automaticamente os workers caso o número mínimo de instâncias não seja garantido.

5.2Configuração do Temma para workers

A configuração do Temma deve conter informações sobre o storage e, quando aplicável, o transport das tarefas.

Exemplo de arquivo etc/temma.php, com armazenamento MySQL:

<?php

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

Outro exemplo de arquivo etc/temma.php, com transport Beanstalkd:

<?php

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

Outro exemplo de arquivo etc/temma.php, com transport Amazon SQS:

<?php

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

Por padrão, os workers de polling aguardam 60 segundos entre duas conexões para obter as tarefas pendentes. Isso se aplica aos workers que se conectam a uma fila Amazon SQS ou a um banco de dados MySQL ou Redis; não se aplica às filas Beanstalkd.

Esse atraso pode ser modificado usando o parâmetro loopDelay na configuração estendida x-asynk: 

<?php

return [
    'application' => [
        'dataSources' => [
            'db'        => 'mysql://user:password@localhost'
        ]
    ],
    'x-asynk' => [
        'storage'   => 'db',
        // atraso de 90 segundos entre duas verificações
        'loopDelay' => 90
    ],
];

5.3Configuração do Supervisor

O uso do Supervisor é opcional, mas pode ser útil para garantir que os workers estejam em execução e que sejam reiniciados caso ocorra algum problema.

Copie o arquivo etc/asynk/supervisor.conf para /etc/supervisor/conf.d/asynk.conf com o seguinte comando: 

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

Em seguida, modifique os seguintes parâmetros no arquivo:

  • command: informe o caminho correto até a raiz do seu projeto.
    Exemplo: command=/path/to/project/bin/comma '\Temma\Cli\Asynk\Worker'
  • numprocs: defina o número de workers a serem executados simultaneamente.
    Exemplo: numprocs=5

Em seguida, force o Supervisor a considerar essa configuração:

sudo supervisorctl reread
sudo supervisorctl update