massivejobs
1.0.0
Biblioteca de código aberto para processamento de fundo simples e distribuído para .NET, alimentado pelo RabbitMQ Message Broker. A documentação, uma vez concluída, estará disponível no site MassiveJobs.net.
Se você não possui uma instalação existente do RabbitMQ, a maneira mais simples é iniciá -lo em um contêiner. O comando a seguir iniciará o RabbitMQ em um contêiner que será removido imediatamente quando parado .
docker run -- rm -- hostname rabbit - test -- name rabbit - test - d - p 15672 : 15672 - p 5672 : 5672 rabbitmq:managementAgora, você poderá acessar a interface do usuário do RabbitMQ em seu navegador em: http: // localhost: 15672 Endereço. Você pode fazer login com o convidado de nome de usuário e o convidado de senha, se quiser monitorar as conexões, filas, etc.
Usaremos o .NET Core 3.1 CLI para esse início rápido, mas você também pode fazê -lo no Visual Studio, com o .NET Core ou com o .NET Framework 4.6.1 ou posterior.
Crie uma pasta para o projeto.
mkdir MassiveJobs.QuickStart
cd MassiveJobs.QuickStartCrie um novo projeto de aplicativo de console.
dotnet new consoleTeste o projeto de andaime.
dotnet run Você deveria ver Hello World! Depois de alguns segundos.
Adicione uma referência de pacote ao MassiveJobs.RabbitMqBroker .
dotnet add package MassiveJobs.RabbitMqBrokerUse seu editor favorito para abrir o programa.cs e insira este código. Os comentários no código devem ser suficientes para lhe dar uma idéia básica do que está acontecendo.
using System ;
using MassiveJobs . Core ;
using MassiveJobs . RabbitMqBroker ;
namespace MassiveJobs . QuickStart
{
/// <summary>
/// This is a "job" class.
/// It will be instantiated every time a message is received, and Perform will be called.
/// It inherits from Job<TJob, TArgs> generic class, where TJob specifies the type of the job,
/// and TArgs specifies the type of the parameter expected by the Perform method.
///
/// In the example below, TArgs is a string, but it can be a custom class with multiple properties.
/// TArgs instances will be serialized (System.Text.Json by default) as a part of the job,
/// before it gets sent to the RabbitMQ.
/// </summary>
public class MessageReceiver : Job < MessageReceiver , string >
{
public override void Perform ( string message )
{
Console . WriteLine ( "Job performed: " + message ) ;
}
}
class Program
{
private static void Main ( )
{
Console . WriteLine ( "1: Worker" ) ;
Console . WriteLine ( "2: Publisher" ) ;
Console . Write ( "Choose 1 or 2 -> " ) ;
var startWorkers = Console . ReadLine ( ) != "2" ;
// We are not starting job workers if '2' is selected.
// This is not mandatory, an application can run job workers
// and publish jobs using the same MassiveJobs instance.
JobsBuilder . Configure ( )
. WithRabbitMqBroker ( )
. Build ( startWorkers ) ;
if ( startWorkers )
{
RunWorker ( ) ;
}
else
{
RunPublisher ( ) ;
}
JobsBuilder . DisposeJobs ( ) ;
}
private static void RunWorker ( )
{
Console . WriteLine ( "Initialized job worker." ) ;
Console . WriteLine ( "Press Enter to end the application." ) ;
Console . ReadLine ( ) ;
}
private static void RunPublisher ( )
{
Console . WriteLine ( "Initialized job publisher" ) ;
Console . WriteLine ( "Write the job name and press Enter to publish it (empty job name to end)." ) ;
while ( true )
{
Console . Write ( "> " ) ;
var message = Console . ReadLine ( ) ;
if ( string . IsNullOrWhiteSpace ( message ) ) break ;
// notice that Publish is a static method on our MessageReceiver class
// it is available because MessageReceiver inherits from Job<TJob, TArgs>
MessageReceiver . Publish ( message ) ;
}
}
}
}Inicie três instruções de comando diferentes (ou conchas de energia). Dois serão usados como trabalhadores e um será usado como editor.
Para iniciar o aplicativo, vá para a pasta do projeto e execute:
dotnet run Digite 1 e pressione Enter para iniciar um trabalhador, tipo 2 e pressionar Enter para iniciar o editor. Ao inserir mensagens no console do editor, você os notará sendo processados em um ou em outro trabalhador, mas não em ambos. Isso ocorre porque os empregos são distribuídos entre os trabalhadores.
Observe que você também pode iniciar vários editores.
Trabalhadores e editores podem estar em diferentes máquinas, desde que possam acessar o servidor RabbitMQ.
Para distribuir os trabalhadores em várias máquinas, você terá que configurar as informações sobre o servidor RabbitMQ. No mínimo, isso significa nome de usuário, senha, nome do host (ou endereço IP) e o número da porta (se o servidor RabbitMQ estiver configurado para ouvir conexões em uma porta não padrão). No exemplo acima, não configuramos nada porque os padrões eram suficientes - nome de usuário: guest , senha: guest , nome do host: localhost , porta: -1 (= use a porta padrão).
Por exemplo, se o seu servidor RabbitMQ estiver em execução em uma máquina com o host Nome rabbit.example.local , ouvindo o número da porta padrão, e você criou um usuário massive no RabbitMqJobs com a senha: d0ntUseTh!sPass
JobsBuilder . Configure ( )
. WithRabbitMqBroker ( s =>
{
s . HostNames = new [ ] { "rabbit.example.com" } ;
s . Username = "massive" ;
s . Password = "d0ntUseTh!sPass" ;
} )
. Build ( ) ;Ou, se você não quiser iniciar os threads do trabalhador (ou seja, usar o processo apenas para publicar trabalhos):
JobsBuilder . Configure ( )
. WithRabbitMqBroker ( s =>
{
//...
} )
. Build ( false ) ;Agora você pode implantar trabalhadores (e editores) em várias máquinas e executá -las. Se a conectividade da rede estiver funcionando (firewalls abertos etc.) tudo deve funcionar. Os empregos seriam encaminhados para os trabalhadores de uma maneira redonda. Lembre -se de que, por padrão, todo aplicativo MassiveJobs está iniciando dois threads de trabalhadores. Isso significa que, se você tiver 3 máquinas, cada uma executando um aplicativo MassiveJobs, a distribuição de empregos se pareceria com isso:
Você deve ter notado, no exemplo rápido, quando tivemos dois aplicativos massivejobs em duas janelas de Posershell, duas das mensagens iriam para uma janela, as próximas duas para a outra janela e assim por diante. Agora você sabe o motivo.
Pule esta seção se o seu aplicativo estiver em execução em um ambiente hospedado .NET Core (APORMA DE WEB CORE ASP.NET Core ou Serviço de Trabalhador).
É muito importante configurar o registro no seu aplicativo executando o MassiveJobs, porque essa é a única maneira de ver erros de tempo de execução do MassiveJobs em seu aplicativo. É tão simples quanto instalar um pacote adequado e definir o JobLoggerFactory na inicialização, se você estiver usando uma das seguintes bibliotecas de logger:
MassiveJobs.Logging.Log4Net )MassiveJobs.Logging.NLog )MassiveJobs.Logging.Serilog ) Por exemplo, se você deseja adicionar log4net ao exemplo do exemplo de partida rápida, primeiro instale o pacote MassiveJobs.Logging.Log4Net em seu projeto. Depois disso, inicialize a biblioteca log4net e, finalmente, massivejobs.
//...
using MassiveJobs . Logging . Log4Net ;
//...
private static void Main ( )
{
InitializeLogging ( ) ;
Console . WriteLine ( "1: Worker" ) ;
Console . WriteLine ( "2: Publisher" ) ;
Console . Write ( "Choose 1 or 2 -> " ) ;
var startWorkers = Console . ReadLine ( ) != "2" ;
// We are not starting job workers if '2' is selected.
// This is not mandatory, an application can run job workers
// and publish jobs using the same MassiveJobs instance.
JobsBuilder . Configure ( )
. WithLog4Net ( )
. WithRabbitMqBroker ( )
. Build ( startWorkers ) ;
if ( startWorkers )
{
RunWorker ( ) ;
}
else
{
RunPublisher ( ) ;
}
}Você precisa implementar a "InicializeLogging", pois normalmente faz a inicialização da sua biblioteca de log. Por exemplo, para o log4net, isso configuraria apenas o Appender do console.
private static void InitializeLogging ( )
{
var patternLayout = new PatternLayout ( ) ;
patternLayout . ConversionPattern = "%date [%thread] %-5level %logger - %message%newline" ;
patternLayout . ActivateOptions ( ) ;
var hierarchy = ( Hierarchy ) LogManager . GetRepository ( Assembly . GetExecutingAssembly ( ) ) ;
hierarchy . Root . AddAppender ( new ConsoleAppender { Layout = patternLayout } ) ;
hierarchy . Root . Level = Level . Warn ;
hierarchy . Configured = true ;
}Agora, quando você inicia o aplicativo do trabalhador, verá as mensagens de registro no console:
PS > dotnet run
1 : Worker
2 : Publisher
Choose 1 or 2 - > 1
2020 - 11 - 10 10 : 25 : 22 , 251 [ 1 ] WARN MassiveJobs.RabbitMqBroker.RabbitMqMessageConsumer - Connected
Initialized job worker.
Press Enter to end the application.Você notará que, se você iniciar o aplicativo do editor, ele não tentará se conectar ao RabbitMQ até tentar enviar as primeiras mensagens. Isso ocorre porque todo aplicativo MassiveJobs mantém duas conexões com o RabbitMQ, uma para publicação e outra para consumir mensagens. No editor, não estamos iniciando trabalhadores, portanto, consumir conexões não é inicializado.
PS > dotnet run
1 : Worker
2 : Publisher
Choose 1 or 2 - > 2
Initialized job publisher
Write the job name and press Enter to publish it (empty job name to end ).
> Hello
2020 - 11 - 10 10 : 27 : 22 , 954 [ 4 ] WARN MassiveJobs.RabbitMqBroker.RabbitMqMessagePublisher - Connected Para usar MassiveJobs.RabbitMqBroker em um ambiente hospedado .NET Core (ASP.NET Core, Serviços de Trabalhador) Instale o seguinte pacote em seu aplicativo:
dotnet add package MassiveJobs.RabbitMqBroker.Hosting Então, na sua aula de inicialização, ao configurar os serviços, ligue para services.AddMassiveJobs() .
//...
using MassiveJobs . RabbitMqBroker . Hosting ;
namespace MassiveJobs . Examples . Api
{
public class Startup
{
public Startup ( IConfiguration configuration )
{
Configuration = configuration ;
}
public IConfiguration Configuration { get ; }
public void ConfigureServices ( IServiceCollection services )
{
//...
services . AddMassiveJobs ( )
. UseRabbitMqBroker ( ) ;
}
//...
}
} Isso registrará os serviços massivejobs necessários e iniciará um serviço hospedado em segundo plano para administrar os trabalhadores. Agora você pode publicar trabalhos de um controlador. Por exemplo, se você tiver uma entidade Customer e deseja enviar um e -mail de boas -vindas para um cliente recém -criado, você pode ter algo assim:
// POST: api/Customers
[ HttpPost ]
public ActionResult < Customer > PostCustomer ( Customer customer )
{
using var trans = _context . Database . BeginTransaction ( ) ;
_context . Customers . Add ( customer ) ;
_context . SaveChanges ( ) ;
if ( ! string . IsNullOrWhiteSpace ( customer . Email ) )
{
// send a welcome email after 5 seconds
SendWelcomeEmailJob . Publish ( customer . Id , TimeSpan . FromSeconds ( 5 ) ) ;
}
// do this last. If Job publishing to RabbitMq fails, we will rollback
trans . Commit ( ) ;
return CreatedAtAction ( "GetCustomer" , new { id = customer . Id } , customer ) ;
} É muito importante ter em mente que SendWelcomeEmailJob.Publish não participa da transação _. O RabbitMQBroker para MassiveJobs não suporta transações. Mas, o método Publish lançará exceção se a publicação falhar (apenas a publicação - na verdade não enviando o e -mail, o que é feito de forma assíncrona). Se a publicação falhar, a exceção será lançada e trans.Commit() nunca será chamado, e a transação será transbordada em disposição.
Emencialmente, a publicação de um trabalho é usada aqui como um último recurso de compromisso .
O SendWelcomeEmailJob poderia parecer algo assim:
public class SendWelcomeEmailJob : Job < SendWelcomeEmailJob , int >
{
private readonly ExamplesDbContext _context ;
public SendWelcomeEmailJob ( ExamplesDbContext context )
{
_context = context ;
}
public override void Perform ( int customerId )
{
using var trans = _context . Database . BeginTransaction ( ) ;
var customer = _context . Customers . Find ( customerId ) ;
if ( customer . IsEmailSent ) return ; // make the job idempotent
customer . IsEmailSent = true ;
// Do this before sending email, to lessen the chance of an exception on commit.
// Also, if optimistic concurrency is enabled, we will fail here, before sending the email.
// This way we avoid sending the email to the customer twice.
_context . SaveChanges ( ) ;
SendEmail ( customer ) ;
// Do this last. In case the SendEmail method fails, the transaction will be rolled back.
trans . Commit ( ) ;
}
private static void SendEmail ( Customer customer )
{
var mailMessage = new MailMessage
{
From = new MailAddress ( "[email protected]" ) ,
Body = $ "Welcome customer { customer . FirstName } { customer . LastName } " ,
Subject = "Welcome to examples.com"
} ;
mailMessage . To . Add ( customer . Email ) ;
using ( var client = new SmtpClient ( "smtp.examples.com" ) )
{
client . UseDefaultCredentials = false ;
client . Credentials = new NetworkCredential ( "username" , "password" ) ;
client . Send ( mailMessage ) ;
}
}
}Há várias coisas a serem observadas aqui:
customer.IsEmailSent é verificado antes de fazer qualquer coisa. Se estiver definido como True, não fazemos nada (nenhuma exceção é lançada, porque a exceção tornaria a biblioteca massivejobs agendar o trabalho para tentativas)SaveChanges() no contexto do banco de dados antes de enviar o email para que ele possa lançar exceções de simultaneidade que reagerem o trabalho para posteriormente (mas você deve configurar propriedades de simultaneidade em seus alunos para que ele funcione). No entanto, nesse caso em particular, nossa classe de trabalho não é totalmente idempotente. Ainda pode acontecer que o email seja enviado duas vezes porque o servidor de email não participa da transação. Se client.Send lança exceção de tempo limite , não se sabe se o email foi realmente enviado ou não. O servidor de email pode ter recebido a solicitação, filmou a mensagem para entrega, mas nunca recebemos a resposta devido a um problema de rede temporário. Em outras palavras, pelo menos uma vez que a entrega é garantida neste caso, não exatamente uma vez .
Se apenas as alterações no banco de dados estivessem envolvidas no trabalho, poderíamos garantir exatamente uma vez . Mas, mesmo assim, o método Perform do trabalho pode ser chamado duas vezes, portanto, você deve garantir que o trabalho seja idempotente no método Perform (semelhante ao que fizemos com IsEmailSent ).
