Vytváření podnikových aplikací řízených zprávami pomocí NServiceBus a Azure Service Bus

Článek
06/21/2024

NServiceBus je komerční architektura zasílání zpráv poskytovaná konkrétním softwarem. Je založená na službě Azure Service Bus a pomáhá vývojářům soustředit se na obchodní logiku tím, že abstrahuje obavy z infrastruktury. V této příručce vytvoříme řešení, které vyměňuje zprávy mezi dvěma službami. Ukážeme si také, jak automaticky opakovat neúspěšné zprávy a zkontrolovat možnosti hostování těchto služeb v Azure.

Poznámka:

Kód pro tento kurz je k dispozici na webu Dokumentace ke konkrétnímu softwaru.

Požadavky

Ukázka předpokládá, že jste vytvořili obor názvů služby Azure Service Bus.

Důležité

NServiceBus vyžaduje alespoň úroveň Standard. Úroveň Basic nebude fungovat.

Stažení a příprava řešení

Stáhněte si kód z webu Dokumentace ke konkrétnímu softwaru. Řešení SendReceiveWithNservicebus.sln se skládá ze tří projektů:
- Odesílatel: konzolová aplikace, která odesílá zprávy
- Příjemce: konzolová aplikace, která přijímá zprávy od odesílatele a odpovídá zpět
- Sdílené: knihovna tříd obsahující kontrakty zpráv sdílené mezi odesílatelem a příjemcem
Následující diagram vygenerovaný službou ServiceInsight, vizualizační a ladicí nástroj z konkrétního softwaru znázorňuje tok zpráv:
Otevřete SendReceiveWithNservicebus.sln v oblíbeném editoru kódu (například Visual Studio 2022).
Otevřete appsettings.json v projektech Příjemce i Odesílatel a nastavte AzureServiceBusConnectionString připojovací řetězec pro váš obor názvů služby Azure Service Bus.
- Najdete ji na webu Azure Portal v části Nastavení>oboru názvů>služby Service Bus Zásady>sdíleného přístupu RootManageSharedAccessKey>Primární připojovací řetězec .
- Má AzureServiceBusTransport také konstruktor, který přijímá přihlašovací údaje k oboru názvů a tokenu, které v produkčním prostředí budou bezpečnější, ale pro účely tohoto kurzu se použije sdílený přístupový klíč připojovací řetězec.

Definování sdílených kontraktů zpráv

Knihovna sdílených tříd je místo, kde definujete kontrakty používané k odesílání našich zpráv. Obsahuje odkaz na NServiceBus balíček NuGet, který obsahuje rozhraní, která můžete použít k identifikaci našich zpráv. Rozhraní nejsou povinná, ale poskytují nám další ověření z NServiceBus a umožňují, aby kód byl samostatně zdokumentovaný.

Nejprve si projdeme Ping.cs třídu.

public class Ping : NServiceBus.ICommand
{
    public int Round { get; set; }
}

Třída Ping definuje zprávu, kterou odesílatel odešle příjemci. Jedná se o jednoduchou třídu jazyka C#, která implementuje NServiceBus.ICommandrozhraní z balíčku NServiceBus. Tato zpráva je signálem pro čtenáře a NServiceBus, že se jedná o příkaz, i když existují další způsoby identifikace zpráv bez použití rozhraní.

Druhá třída zprávy ve sdílených projektech je Pong.cs:

public class Pong : NServiceBus.IMessage
{
    public string Acknowledgement { get; set; }
}

Pong je také jednoduchý objekt jazyka C#, i když tento objekt implementuje NServiceBus.IMessage. Rozhraní IMessage představuje obecnou zprávu, která není příkazem ani událostí a běžně se používá pro odpovědi. V naší ukázce je to odpověď, kterou příjemce pošle zpět odesílateli, aby označil, že zpráva byla přijata.

Pong Jedná se o Ping dva typy zpráv, které použijete. Dalším krokem je konfigurace odesílatele pro použití služby Azure Service Bus a odeslání Ping zprávy.

Nastavení odesílatele

Odesílatel je koncový bod, který odesílá naši Ping zprávu. Tady nakonfigurujete odesílatele tak, aby jako transportní mechanismus používal Službu Azure Service Bus, a pak vytvořil Ping instanci a odeslal ji.

Main V metodě Program.csnakonfigurujete koncový bod odesílatele:

var host = Host.CreateDefaultBuilder(args)
    // Configure a host for the endpoint
    .ConfigureLogging((context, logging) =>
    {
        logging.AddConfiguration(context.Configuration.GetSection("Logging"));

        logging.AddConsole();
    })
    .UseConsoleLifetime()
    .UseNServiceBus(context =>
    {
        // Configure the NServiceBus endpoint
        var endpointConfiguration = new EndpointConfiguration("Sender");

        var connectionString = context.Configuration.GetConnectionString("AzureServiceBusConnectionString");
        // If token credentials are to be used, the overload constructor for AzureServiceBusTransport would be used here
        var routing = endpointConfiguration.UseTransport(new AzureServiceBusTransport(connectionString));
        endpointConfiguration.UseSerialization<SystemJsonSerializer>();

        endpointConfiguration.AuditProcessedMessagesTo("audit");
        routing.RouteToEndpoint(typeof(Ping), "Receiver");

        endpointConfiguration.EnableInstallers();

        return endpointConfiguration;
    })
    .ConfigureServices(services => services.AddHostedService<SenderWorker>())
    .Build();

await host.RunAsync();

Tady je hodně věcí rozbalit, takže si ho projdeme krok za krokem.

Konfigurace hostitele pro koncový bod

Hostování a protokolování se konfiguruje pomocí standardních možností Microsoft Generic Host. Prozatím je koncový bod nakonfigurovaný tak, aby běžel jako konzolová aplikace, ale dá se upravit tak, aby běžel ve službě Azure Functions s minimálními změnami, které probereme dále v tomto článku.

Konfigurace koncového bodu NServiceBus

Dále řeknete hostiteli, aby používal NServiceBus s rozšiřující metodou .UseNServiceBus(…) . Metoda přebírá funkci zpětného volání, která vrací koncový bod, který se spustí při spuštění hostitele.

V konfiguraci koncového bodu zadáte AzureServiceBus pro přenos připojovací řetězec z appsettings.json. Dále nastavíte směrování tak, aby se zprávy typu Ping odesílaly do koncového bodu s názvem Příjemce. Umožňuje službě NServiceBus automatizovat proces odesílání zprávy do cíle, aniž by vyžadoval adresu příjemce.

EnableInstallers Volání nastaví naši topologii v oboru názvů služby Azure Service Bus při spuštění koncového bodu a v případě potřeby vytvoří požadované fronty. V produkčních prostředích je provozní skriptování další možností vytvoření topologie.

Nastavení služby na pozadí pro odesílání zpráv

Poslední částí odesílatele je SenderWorkerslužba na pozadí, která je nakonfigurovaná tak, aby posílala zprávu každou sekundu Ping .

public class SenderWorker : BackgroundService
{
    private readonly IMessageSession messageSession;
    private readonly ILogger<SenderWorker> logger;

    public SenderWorker(IMessageSession messageSession, ILogger<SenderWorker> logger)
    {
        this.messageSession = messageSession;
        this.logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            var round = 0;
            while (!stoppingToken.IsCancellationRequested)
            {
                await messageSession.Send(new Ping { Round = round++ });;

                logger.LogInformation($"Message #{round}");

                await Task.Delay(1_000, stoppingToken);
            }
        }
        catch (OperationCanceledException)
        {
            // graceful shutdown
        }
    }
}

Použitý IMessageSession injektáž ExecuteAsync se vloží do SenderWorker a umožňuje nám odesílat zprávy pomocí NServiceBus mimo obslužnou rutinu zprávy. Směrování, které jste nakonfigurovali, Sender určuje cíl Ping zpráv. Udržuje topologii systému (které zprávy jsou směrovány na které adresy) jsou oddělené od obchodního kódu.

Aplikace Sender obsahuje také .PongHandler Vrátíte se k němu poté, co probereme příjemce, který uděláme v dalším kroku.

Nastavení přijímače

Příjemce je koncový bod, který naslouchá Ping zprávě, protokoluje při přijetí zprávy a odpovídá zpět odesílateli. V této části rychle zkontrolujeme konfiguraci koncového bodu, která je podobná odesílateli, a pak obrátíme pozornost na obslužnou rutinu zprávy.

Stejně jako odesílatel nastavte příjemce jako konzolovou aplikaci pomocí nástroje Microsoft Generic Host. Používá stejnou konfiguraci protokolování a koncového bodu (s Azure Service Bus jako přenos zpráv), ale s jiným názvem, aby se odlišila od odesílatele:

var endpointConfiguration = new EndpointConfiguration("Receiver");

Vzhledem k tomu, že tento koncový bod odpoví pouze na jeho původce a nespustí nové konverzace, není nutná žádná konfigurace směrování. Nepotřebuje také pracovníka na pozadí, jako je odesílatel, protože odpoví pouze při přijetí zprávy.

Obslužná rutina zprávy ping

Projekt Receiver obsahuje obslužnou rutinu zprávy s názvemPingHandler:

public class PingHandler : NServiceBus.IHandleMessages<Ping>
{
    private readonly ILogger<PingHandler> logger;

    public PingHandler(ILogger<PingHandler> logger)
    {
        this.logger = logger;
    }

    public async Task Handle(Ping message, IMessageHandlerContext context)
    {
        logger.LogInformation($"Processing Ping message #{message.Round}");

        // throw new Exception("BOOM");

        var reply = new Pong { Acknowledgement = $"Ping #{message.Round} processed at {DateTimeOffset.UtcNow:s}" };

        await context.Reply(reply);
    }
}

Prozatím ignorujme komentářovaný kód; Později se k němu vrátíme, až budeme mluvit o zotavení z selhání.

Třída implementuje IHandleMessages<Ping>, která definuje jednu metodu: Handle. Toto rozhraní říká NServiceBus, že když koncový bod obdrží zprávu typu Ping, měla by být zpracována metodou Handle v této obslužné rutině. Metoda Handle přebírá samotnou zprávu jako parametr a , IMessageHandlerContextkterá umožňuje další operace zasílání zpráv, jako je odpověď, odesílání příkazů nebo publikování událostí.

Naše PingHandler je jednoduchá: při Ping přijetí zprávy zapište podrobnosti zprávy a odpovězte odesílateli novou zprávou Pong , která se následně zpracuje v odesílateli PongHandler.

Poznámka:

V konfiguraci odesílatele jste určili, že Ping zprávy by se měly směrovat do příjemce. NServiceBus přidává metadata do zpráv označujících mimo jiné původ zprávy. Proto nemusíte zadávat žádná směrovací data pro Pong zprávu odpovědi. Automaticky se směruje zpět do jejího původu: Odesílatel.

S správně nakonfigurovaným odesílatelem i příjemcem teď můžete řešení spustit.

Spuštění řešení

Pokud chcete řešení spustit, musíte spustit odesílatele i příjemce. Pokud používáte Visual Studio Code, spusťte konfiguraci Ladit vše. Pokud používáte Visual Studio, nakonfigurujte řešení tak, aby spouštěla projekty Sender i Receiver:

Klikněte pravým tlačítkem na řešení v Průzkumník řešení
Vyberte Nastavit spouštěné projekty...
Výběr více projektů po spuštění
U odesílatele i příjemce vyberte v rozevíracím seznamu možnost Spustit.

Spusťte řešení. Zobrazí se dvě konzolové aplikace, jedna pro odesílatele a jedna pro příjemce.

Odesílatel
Přijímač

Obrázek znázorňující koncový bod odesílající zprávy Ping příjemci

V odesílateli si všimněte, že Ping se zpráva odesílá každou sekundu SenderWorker , a to díky úloze na pozadí. Příjemce zobrazí podrobnosti o každé Ping zprávě, kterou obdrží, a odesílatel zaprokoluje podrobnosti o každé Pong zprávě, kterou obdrží v odpovědi.

Teď, když máte všechno, co funguje, pojďme si ho zlomit.

Odolnost v akci

Chyby jsou faktem života v softwarových systémech. Je nevyhnutelné, že kód selže a může to udělat z různých důvodů, jako jsou selhání sítě, zámky databáze, změny v rozhraní API třetí strany a běžné staré chyby kódování.

NServiceBus má robustní funkce obnovitelnosti pro zpracování chyb. Když obslužná rutina zprávy selže, zprávy se automaticky zopakují na základě předdefinovaných zásad. Existují dva typy zásad opakování: okamžité opakování a zpožděné opakování. Nejlepším způsobem, jak popsat, jak fungují, je vidět je v akci. Pojďme do koncového bodu příjemce přidat zásadu opakování:

Otevřít Program.cs v projektu Odesílatel
.EnableInstallers Za řádek přidejte následující kód:

endpointConfiguration.SendFailedMessagesTo("error");
var recoverability = endpointConfiguration.Recoverability();
recoverability.Immediate(
    immediate =>
    {
        immediate.NumberOfRetries(3);
    });
recoverability.Delayed(
    delayed =>
    {
        delayed.NumberOfRetries(2);
        delayed.TimeIncrease(TimeSpan.FromSeconds(5));
    });

Než probereme, jak tato zásada funguje, podívejme se na ni v praxi. Před otestování zásad obnovitelnosti je potřeba simulovat chybu. PingHandler Otevřete kód v projektu Receiver a odkomentujte tento řádek:

throw new Exception("BOOM");

Když příjemce zpracuje Ping zprávu, selže. Znovu spusťte řešení a pojďme se podívat, co se stane v příjemci.

S naší méně spolehlivou PingHandler, všechny naše zprávy selžou. Zobrazí se zásady opakování, které se u těchto zpráv spustí. Při prvním selhání zprávy se okamžitě opakuje až třikrát:

Obrázek znázorňující okamžitou zásadu opakování, která opakuje zprávy až 3krát

Samozřejmě to bude i nadále selhat, takže když se použijí tři okamžité opakování, zpožděné zásady opakování se spustí a zpráva se zpozdí po dobu 5 sekund:

Obrázek znázorňující zpožděné zásady opakování, které zpožďují zprávy v přírůstcích po 5 sekundách před pokusem o další kolo okamžitého opakování

Po uplynutí těchto 5 sekund se zpráva znovu opakuje třikrát (to znamená další iterace zásad okamžitého opakování). Tyto příkazy také selžou a služba NServiceBus zpozdí zprávu znovu, tentokrát na 10 sekund, než to zkusíte znovu.

Pokud PingHandler se i po spuštění úplné zásady opakování nepodaří, zpráva se umístí do centralizované fronty chyb s názvem error, jak je definováno voláním SendFailedMessagesTo.

Obrázek znázorňující zprávu, která selhala

Koncept centralizované fronty chyb se liší od mechanismu nedoručených zpráv ve službě Azure Service Bus, který má frontu nedoručených zpráv pro každou frontu zpracování. Fronty nedoručených zpráv ve službě Azure Service Bus fungují v případě NServiceBus jako skutečné jedovaté fronty zpráv, zatímco zprávy, které končí v centralizované frontě chyb, je možné v případě potřeby později znovu zpracovat.

Zásady opakování pomáhají řešit několik typů chyb , které jsou často přechodné nebo polo přechodné povahy. To znamená, že chyby, které jsou dočasné a často zmizí, pokud je zpráva jednoduše znovu zpracována po krátké prodlevě. Mezi příklady patří selhání sítě, zámky databází a výpadky rozhraní API třetích stran.

Jakmile je zpráva ve frontě chyb, můžete si prohlédnout podrobnosti zprávy v nástroji podle vašeho výběru a pak se rozhodnout, co s ní dělat. Například pomocí servicePulse, monitorovacího nástroje podle konkrétního softwaru, můžeme zobrazit podrobnosti zprávy a důvod selhání:

Obrázek znázorňující ServicePulse z konkrétního softwaru

Po prozkoumání podrobností můžete zprávu odeslat zpět do původní fronty ke zpracování. Než to uděláte, můžete zprávu také upravit. Pokud je ve frontě chyb více zpráv, které selhaly ze stejného důvodu, můžou být všechny odeslány zpět do původních cílů jako dávka.

V dalším kroku je čas zjistit, kde nasadit naše řešení v Azure.

Kde hostovat služby v Azure

V této ukázce jsou koncové body odesílatele a příjemce nakonfigurované tak, aby běžely jako konzolové aplikace. Dají se také hostovat v různých službách Azure, včetně Azure Functions, Aplikace Azure Services, Azure Container Instances, Azure Kubernetes Services a virtuálních počítačů Azure. Tady je příklad, jak můžete koncový bod odesílatele nakonfigurovat tak, aby běžel jako funkce Azure:

[assembly: NServiceBusTriggerFunction("Sender")]
public class Program
{
    public static async Task Main()
    {
        var host = new HostBuilder()
            .ConfigureFunctionsWorkerDefaults()
            .UseNServiceBus(configuration =>
            {
                configuration.Routing().RouteToEndpoint(typeof(Ping), "Receiver");
            })
            .Build();

        await host.RunAsync();
    }
}

Další informace o používání služby NServiceBus se službou Functions najdete v dokumentaci ke službě NServiceBus v azure Functions se službou Azure Service Bus .

Další kroky

Další informace o používání služby NServiceBus se službami Azure najdete v následujících článcích:

Sdílet prostřednictvím