Delen via

Abonneren op gebeurtenissen

Aanbeveling

Deze inhoud is een fragment uit het eBook, .NET Microservices Architecture for Containerized .NET Applications, beschikbaar op .NET Docs of als een gratis downloadbare PDF die offline kan worden gelezen.

PDF downloaden

.NET Microservices Architectuur voor Gecontaineriseerde .NET Toepassingen eBook omslagthumbnail.

De eerste stap voor het gebruik van de event-bus is het aanmelden van microservices voor de gebeurtenissen die ze willen ontvangen. Deze functionaliteit moet worden uitgevoerd in de ontvangermicroservices.

De volgende eenvoudige code laat zien wat elke ontvanger microservice moet implementeren bij het starten van de service (dat wil gezegd, in de Startup klasse), zodat deze zich abonneert op de gebeurtenissen die het nodig heeft. In dit geval moet de basket-api microservice zich abonneren op ProductPriceChangedIntegrationEvent en de OrderStartedIntegrationEvent berichten.

Wanneer u zich bijvoorbeeld abonneert op het ProductPriceChangedIntegrationEvent event, wordt de winkelwagen microservice op de hoogte gesteld van eventuele wijzigingen in de productprijs en kan het de gebruiker waarschuwen voor de verandering als dat product zich in het mandje van de gebruiker bevindt.

var eventBus = app.ApplicationServices.GetRequiredService<IEventBus>();

eventBus.Subscribe<ProductPriceChangedIntegrationEvent,
                   ProductPriceChangedIntegrationEventHandler>();

eventBus.Subscribe<OrderStartedIntegrationEvent,
                   OrderStartedIntegrationEventHandler>();

Nadat deze code is uitgevoerd, luistert de abonnementsmicroservice via RabbitMQ-kanalen. Wanneer een bericht van het type ProductPriceChangedIntegrationEvent binnenkomt, roept de code de gebeurtenis-handler aan die eraan wordt doorgegeven en verwerkt de gebeurtenis.

Gebeurtenissen publiceren via de gebeurtenisbus

Ten slotte publiceert de afzender van het bericht (oorspronkelijke microservice) de integratie-gebeurtenissen met code die vergelijkbaar is met het volgende voorbeeld. (Deze benadering is een vereenvoudigd voorbeeld dat geen rekening houdt met atomiciteit.) U implementeert vergelijkbare code wanneer een gebeurtenis moet worden doorgegeven aan meerdere microservices, meestal direct na het doorvoeren van gegevens of transacties van de oorspronkelijke microservice.

Eerst zou het event bus-implementatieobject (op basis van RabbitMQ of op een servicebus) worden geïnjecteerd in de controllerconstructor, zoals in de volgende code:

[Route("api/v1/[controller]")]
public class CatalogController : ControllerBase
{
    private readonly CatalogContext _context;
    private readonly IOptionsSnapshot<Settings> _settings;
    private readonly IEventBus _eventBus;

    public CatalogController(CatalogContext context,
        IOptionsSnapshot<Settings> settings,
        IEventBus eventBus)
    {
        _context = context;
        _settings = settings;
        _eventBus = eventBus;
    }
    // ...
}

Vervolgens gebruikt u deze op basis van de methoden van uw controller, zoals in de Methode UpdateProduct:

[Route("items")]
[HttpPost]
public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem product)
{
    var item = await _context.CatalogItems.SingleOrDefaultAsync(
        i => i.Id == product.Id);
    // ...
    if (item.Price != product.Price)
    {
        var oldPrice = item.Price;
        item.Price = product.Price;
        _context.CatalogItems.Update(item);
        var @event = new ProductPriceChangedIntegrationEvent(item.Id,
            item.Price,
            oldPrice);
        // Commit changes in original transaction
        await _context.SaveChangesAsync();
        // Publish integration event to the event bus
        // (RabbitMQ or a service bus underneath)
        _eventBus.Publish(@event);
        // ...
    }
    // ...
}

Omdat de oorspronkelijke microservice een eenvoudige CRUD-microservice is, wordt die code rechtstreeks in een web-API-controller geplaatst.

In geavanceerdere microservices, zoals bij het gebruik van CQRS-benaderingen, kan deze worden geïmplementeerd in de CommandHandler klasse, binnen de Handle() methode.

Atomiciteit en veerkracht ontwerpen bij het publiceren naar de eventbus

Wanneer u integratiegebeurtenissen publiceert via een gedistribueerd berichtensysteem zoals uw gebeurtenisbus, hebt u het probleem van het atomisch bijwerken van de oorspronkelijke database en het publiceren van een gebeurtenis (dat wil gezegd, beide bewerkingen zijn voltooid of geen van deze). In het vereenvoudigde voorbeeld dat eerder wordt weergegeven, worden bijvoorbeeld gegevens door de code doorgevoerd in de database wanneer de productprijs wordt gewijzigd en vervolgens een ProductPriceChangedIntegrationEvent-bericht gepubliceerd. In eerste instantie kan het er essentieel uitzien dat deze twee bewerkingen atomisch worden uitgevoerd. Als u echter een gedistribueerde transactie gebruikt met betrekking tot de database en de berichtenbroker, zoals in oudere systemen zoals Microsoft Message Queuing (MSMQ), wordt deze benadering niet aanbevolen om de redenen die worden beschreven door de CAP-theorema.

In principe gebruikt u microservices om schaalbare en maximaal beschikbare systemen te bouwen. Enigszins vereenvoudigen, zegt de CAP-theorema dat u geen (gedistribueerde) database (of een microservice die eigenaar is van het model) kunt bouwen die voortdurend beschikbaar, sterk consistent en tolerant is voor elke partitie. U moet twee van deze drie eigenschappen kiezen.

In op microservices gebaseerde architecturen moet u beschikbaarheid en tolerantie kiezen en de nadruk op sterke consistentie verminderen. Daarom wilt u in de meeste moderne microservicetoepassingen meestal geen gedistribueerde transacties gebruiken in berichten, net als wanneer u gedistribueerde transacties implementeert op basis van de Windows Distributed Transaction Coordinator (DTC) met MSMQ.

Laten we teruggaan naar het eerste probleem en het voorbeeld. Als de service vastloopt nadat de database is bijgewerkt (in dit geval direct na de coderegel met _context.SaveChangesAsync()), maar voordat de integratiegebeurtenis wordt gepubliceerd, kan het algehele systeem inconsistent worden. Deze benadering kan bedrijfskritiek zijn, afhankelijk van de specifieke bedrijfsbewerking waarmee u te maken hebt.

Zoals eerder vermeld in de sectie Architectuur, kunt u verschillende benaderingen hebben voor het oplossen van dit probleem:

  • Het gebruiken van het volledige Event Sourcing patroon.

  • Transactielogboekanalyse gebruiken.

  • Met behulp van het Outbox-patroon. Dit is een transactionele tabel voor het opslaan van de integratie-gebeurtenissen (uitbreiding van de lokale transactie).

Voor dit scenario is het gebruik van het volledige ES-patroon (Event Sourcing) een van de beste benaderingen, als dat niet het beste is. In veel toepassingsscenario's kunt u echter mogelijk geen volledig ES-systeem implementeren. ES betekent dat alleen domeinevenementen in uw transactionele database worden opgeslagen in plaats van de huidige statusgegevens op te slaan. Het opslaan van alleen domeingebeurtenissen kan grote voordelen hebben, zoals de geschiedenis van uw systeem beschikbaar hebben en de status van uw systeem op elk moment in het verleden kunnen bepalen. Als u echter een volledig ES-systeem implementeert, moet u het grootste deel van uw systeem opnieuw ontwerpen en vele andere complexiteiten en vereisten introduceert. U wilt bijvoorbeeld een database gebruiken die speciaal is gemaakt voor gebeurtenisbronnen, zoals Event Store of een documentgerichte database zoals Azure Cosmos DB, MongoDB, Cassandra, CouchDB of RavenDB. Event sourcing is een uitstekende aanpak voor dit probleem, maar niet de eenvoudigste oplossing, tenzij u al bekend bent met event sourcing.

De optie voor het gebruik van transactielogboekanalyse ziet er in eerste instantie transparant uit. Als u deze methode echter wilt gebruiken, moet de microservice worden gekoppeld aan uw RDBMS-transactielogboek, zoals het SQL Server-transactielogboek. Deze aanpak is waarschijnlijk niet wenselijk. Een ander nadeel is dat de updates op laag niveau die zijn vastgelegd in het transactielogboek mogelijk niet op hetzelfde niveau zijn als uw integratiegebeurtenissen op hoog niveau. Zo ja, dan kan het proces van reverse-engineering van deze transactielogboekbewerkingen lastig zijn.

Een evenwichtige benadering is een combinatie van een transactionele databasetabel en een vereenvoudigd ES-patroon. U kunt een status gebruiken, zoals 'gereed om de gebeurtenis te publiceren', die u in de oorspronkelijke gebeurtenis hebt ingesteld wanneer u deze doorvoert in de tabel met integratieevenementen. Vervolgens probeert u de gebeurtenis naar de evenementbus te publiceren. Als de actie Gebeurtenis publiceren slaagt, start u een andere transactie in de oorspronkelijke service en verplaatst u de status van 'Gereed om de gebeurtenis te publiceren' naar 'gebeurtenis die al is gepubliceerd'.

Als de actie publiceren-gebeurtenis in de gebeurtenisbus mislukt, zijn de gegevens nog steeds niet inconsistent binnen de oorspronkelijke microservice. De actie is nog steeds gemarkeerd als 'gereed om de gebeurtenis te publiceren' en voor de rest van de services is deze uiteindelijk consistent. U kunt altijd achtergrondtaken hebben die de status van de transacties of integratie-gebeurtenissen controleren. Als de taak een gebeurtenis vindt met de status Gereed om de gebeurtenis te publiceren, kan deze proberen om die gebeurtenis opnieuw te publiceren naar de gebeurtenisbus.

U ziet dat u met deze benadering alleen de integratie-gebeurtenissen voor elke microservice van oorsprong behoudt en alleen de gebeurtenissen die u met andere microservices of externe systemen wilt communiceren. In een volledig ES-systeem slaat u daarentegen ook alle domeinevenementen op.

Daarom is deze evenwichtige benadering een vereenvoudigd ES-systeem. U hebt een lijst met integratie-gebeurtenissen met hun huidige status nodig ('gereed om te publiceren' versus 'gepubliceerd'). Maar u hoeft deze statussen alleen te implementeren voor de integratie-gebeurtenissen. En in deze benadering hoeft u niet al uw domeingegevens op te slaan als gebeurtenissen in de transactionele database, net als in een volledig ES-systeem.

Als u al een relationele database gebruikt, kunt u een transactionele tabel gebruiken om integratie-gebeurtenissen op te slaan. Als u atomiciteit in uw toepassing wilt bereiken, gebruikt u een proces in twee stappen op basis van lokale transacties. In principe hebt u een IntegrationEvent-tabel in dezelfde database waarin u uw domeinentiteiten hebt. Deze tabel fungeert als een garantie voor het bereiken van atomiciteit, zodat u blijvende integratiegebeurtenissen kunt opnemen in dezelfde transacties die uw domeingegevens vastleggen.

Stap voor stap gaat het proces als volgt:

  1. De toepassing begint met een lokale databasetransactie.

  2. Vervolgens wordt de status van uw domeinentiteiten bijgewerkt en wordt een gebeurtenis in de integratiegebeurtenistabel ingevoegd.

  3. Ten slotte wordt de transactie vastgelegd, zodat u de gewenste atomiciteit krijgt.

  4. U publiceert de gebeurtenis op een of andere manier (volgende).

Bij het implementeren van de stappen voor het publiceren van de gebeurtenissen hebt u de volgende opties:

  • Publiceer de integratiegebeurtenis direct na het doorvoeren van de transactie en gebruik een andere lokale transactie om de gebeurtenissen in de tabel te markeren als gepubliceerd. Gebruik vervolgens de tabel net als een artefact om de integratie-gebeurtenissen bij te houden in het geval van problemen in de externe microservices en voer compenserende acties uit op basis van de opgeslagen integratie-gebeurtenissen.

  • Gebruik de tabel als een soort wachtrij. Een afzonderlijke toepassingsthread of -proces voert query's uit op de integratie gebeurtenistabel, publiceert de gebeurtenissen naar de gebeurtenisbus en gebruikt vervolgens een lokale transactie om de gebeurtenissen te markeren als gepubliceerd.

Afbeelding 6-22 toont de architectuur voor de eerste van deze benaderingen.

Diagram van atomiciteit bij het publiceren zonder een werknemer microservice.

Afbeelding 6-22. Atomiciteit bij het publiceren van gebeurtenissen naar de eventbus

In de in afbeelding 6-22 weergegeven aanpak ontbreekt een extra werknemers-microservice die verantwoordelijk is voor het controleren en bevestigen van het succes van de gepubliceerde integratiegebeurtenissen. In het geval van een fout kan de extra checker worker-microservice gebeurtenissen uit de tabel lezen en deze opnieuw publiceren, dat wil gezegd, stap 2 herhalen.

Over de tweede benadering: u gebruikt de EventLog-tabel als wachtrij en gebruikt altijd een werkrolmicroservice om de berichten te publiceren. In dat geval ziet het proces er ongeveer als volgt uit in afbeelding 6-23. Dit toont een extra microservice en de tabel is de enige bron bij het publiceren van gebeurtenissen.

Diagram van atomiciteit bij publicatie met een werkermicroservice.

Afbeelding 6-23. Atomiciteit bij het publiceren van gebeurtenissen naar de eventbus met een worker-microservice

Ter vereenvoudiging gebruikt het voorbeeld eShopOnContainers de eerste benadering (zonder extra processen of controlemicroservices) plus de gebeurtenisbus. Het voorbeeld eShopOnContainers verwerkt echter niet alle mogelijke foutcases. In een echte toepassing die in de cloud is geïmplementeerd, moet u het feit omarmen dat er uiteindelijk problemen optreden en moet u die controle- en opnieuw verzendende logica implementeren. Het gebruik van de tabel als een wachtrij kan effectiever zijn dan de eerste benadering als u deze tabel als een enkele bron van evenementen hebt bij het publiceren ervan (met de worker) via de evenementbus.

Atomiciteit implementeren bij het publiceren van integratiegebeurtenissen via de eventbus

In de volgende code ziet u hoe u één transactie met meerdere DbContext-objecten kunt maken: één context die is gerelateerd aan de oorspronkelijke gegevens die worden bijgewerkt en de tweede context die is gerelateerd aan de tabel IntegrationEventLog.

De transactie in de onderstaande voorbeeldcode is niet tolerant als verbindingen met de database een probleem hebben op het moment dat de code wordt uitgevoerd. Dit kan gebeuren in cloudsystemen zoals Azure SQL DB, waardoor databases mogelijk worden verplaatst tussen servers. Zie de sectie Resilient Entity Framework Core SQL-verbindingen implementeren verderop in deze handleiding voor het implementeren van flexibele transacties in meerdere contexten.

Voor de duidelijkheid toont het volgende voorbeeld het hele proces in één stuk code. De implementatie van eShopOnContainers wordt echter geherstructureerd en splitst deze logica in meerdere klassen, zodat het eenvoudiger te onderhouden is.

// Update Product from the Catalog microservice
//
public async Task<IActionResult> UpdateProduct([FromBody]CatalogItem productToUpdate)
{
  var catalogItem =
       await _catalogContext.CatalogItems.SingleOrDefaultAsync(i => i.Id ==
                                                               productToUpdate.Id);
  if (catalogItem == null) return NotFound();

  bool raiseProductPriceChangedEvent = false;
  IntegrationEvent priceChangedEvent = null;

  if (catalogItem.Price != productToUpdate.Price)
          raiseProductPriceChangedEvent = true;

  if (raiseProductPriceChangedEvent) // Create event if price has changed
  {
      var oldPrice = catalogItem.Price;
      priceChangedEvent = new ProductPriceChangedIntegrationEvent(catalogItem.Id,
                                                                  productToUpdate.Price,
                                                                  oldPrice);
  }
  // Update current product
  catalogItem = productToUpdate;

  // Just save the updated product if the Product's Price hasn't changed.
  if (!raiseProductPriceChangedEvent)
  {
      await _catalogContext.SaveChangesAsync();
  }
  else  // Publish to event bus only if product price changed
  {
        // Achieving atomicity between original DB and the IntegrationEventLog
        // with a local transaction
        using (var transaction = _catalogContext.Database.BeginTransaction())
        {
           _catalogContext.CatalogItems.Update(catalogItem);
           await _catalogContext.SaveChangesAsync();

           await _integrationEventLogService.SaveEventAsync(priceChangedEvent);

           transaction.Commit();
        }

      // Publish the integration event through the event bus
      _eventBus.Publish(priceChangedEvent);

      _integrationEventLogService.MarkEventAsPublishedAsync(
                                                priceChangedEvent);
  }

  return Ok();
}

Nadat de integratiegebeurtenis ProductPriceChangedIntegrationEvent is gemaakt, bevat de transactie die de oorspronkelijke domeinbewerking opslaat (het catalogusitem bijwerken) ook de persistentie van de gebeurtenis in de Tabel EventLog. Dit maakt het één transactie en u kunt altijd controleren of gebeurtenisberichten zijn verzonden.

De gebeurtenislogboektabel wordt atomisch bijgewerkt met de oorspronkelijke databasebewerking, met behulp van een lokale transactie voor dezelfde database. Als een van de bewerkingen mislukt, wordt er een uitzondering gegenereerd en wordt een voltooide bewerking teruggedraaid, waardoor de consistentie tussen de domeinbewerkingen en de gebeurtenisberichten die zijn opgeslagen in de tabel behouden blijft.

Berichten uit abonnementen ontvangen: event-handlers in ontvangende microservices

Naast de logica van het evenementabonnement dient u de interne code te implementeren voor de gebeurtenis-handlers voor integratie, zoals een callbackmethode. De gebeurtenis-handler is waar u opgeeft waar de gebeurtenisberichten van een bepaald type worden ontvangen en verwerkt.

Een event handler ontvangt eerst een gebeurtenisinstantie van de gebeurtenisbus. Vervolgens wordt het onderdeel gevonden dat moet worden verwerkt met betrekking tot die integratiegebeurtenis, het doorgeven en persistent maken van de gebeurtenis als een wijziging in de status van de ontvangermicroservice. Als bijvoorbeeld een ProductPriceChanged-gebeurtenis afkomstig is uit de catalogus-microservice, wordt deze verwerkt in de winkelmandmicroservice en wijzigt de status ook in deze ontvangende winkelmandmicroservice, zoals wordt weergegeven in de volgende code.

namespace Microsoft.eShopOnContainers.Services.Basket.API.IntegrationEvents.EventHandling
{
    public class ProductPriceChangedIntegrationEventHandler :
        IIntegrationEventHandler<ProductPriceChangedIntegrationEvent>
    {
        private readonly IBasketRepository _repository;

        public ProductPriceChangedIntegrationEventHandler(
            IBasketRepository repository)
        {
            _repository = repository;
        }

        public async Task Handle(ProductPriceChangedIntegrationEvent @event)
        {
            var userIds = await _repository.GetUsers();
            foreach (var id in userIds)
            {
                var basket = await _repository.GetBasket(id);
                await UpdatePriceInBasketItems(@event.ProductId, @event.NewPrice, basket);
            }
        }

        private async Task UpdatePriceInBasketItems(int productId, decimal newPrice,
            CustomerBasket basket)
        {
            var itemsToUpdate = basket?.Items?.Where(x => int.Parse(x.ProductId) ==
                productId).ToList();
            if (itemsToUpdate != null)
            {
                foreach (var item in itemsToUpdate)
                {
                    if(item.UnitPrice != newPrice)
                    {
                        var originalPrice = item.UnitPrice;
                        item.UnitPrice = newPrice;
                        item.OldUnitPrice = originalPrice;
                    }
                }
                await _repository.UpdateBasket(basket);
            }
        }
    }
}

De event handler moet nagaan of het product zich bevindt in een van de winkelwageninstanties. De artikelprijs wordt ook bijgewerkt voor elk gerelateerde winkelwagenlijnitem. Ten slotte wordt er een waarschuwing gemaakt die aan de gebruiker moet worden weergegeven over de prijswijziging, zoals weergegeven in afbeelding 6-24.

Schermopname van een browser met de melding prijswijziging op de gebruikerskar.

Afbeelding 6-24. Een prijswijziging van een artikel weergeven in een winkelmandje, zoals gecommuniceerd via integratie-evenementen.

Idempotentie in updatebericht gebeurtenissen

Een belangrijk aspect van updatebericht gebeurtenissen is dat een fout op elk moment in de communicatie ertoe moet leiden dat het bericht opnieuw wordt geprobeerd. Anders kan een achtergrondtaak proberen een evenement te publiceren dat al is gepubliceerd, waardoor een race conditie ontstaat. Zorg ervoor dat de updates idempotent zijn of dat ze voldoende informatie bieden om ervoor te zorgen dat u een duplicaat kunt detecteren, verwijderen en slechts één antwoord kunt terugsturen.

Zoals eerder vermeld, betekent idempotentie dat een bewerking meerdere keren kan worden uitgevoerd zonder het resultaat te wijzigen. In een berichtenomgeving, net als bij het communiceren van gebeurtenissen, is een gebeurtenis idempotent als deze meerdere keren kan worden geleverd zonder het resultaat voor de ontvanger microservice te wijzigen. Dit kan nodig zijn vanwege de aard van de gebeurtenis zelf of vanwege de manier waarop het systeem de gebeurtenis afhandelt. Bericht-idempotentie is belangrijk in elke toepassing die gebruikmaakt van berichten, niet alleen in toepassingen die het event bus-patroon implementeren.

Een voorbeeld van een idempotente bewerking is een SQL-instructie die alleen gegevens in een tabel invoegt als die gegevens zich nog niet in de tabel bevinden. Het maakt niet uit hoe vaak u die SQL-instructie invoegt; het resultaat is hetzelfde: de tabel bevat die gegevens. Idempotentie zoals deze kan ook nodig zijn bij het verwerken van berichten als de berichten mogelijk kunnen worden verzonden en daarom meerdere keren kunnen worden verwerkt. Als de logica voor opnieuw proberen bijvoorbeeld ervoor zorgt dat een afzender hetzelfde bericht meer dan één keer verzendt, moet u ervoor zorgen dat het idempotent is.

Het is mogelijk om idempotente berichten te ontwerpen. U kunt bijvoorbeeld een gebeurtenis maken met de tekst 'Stel de productprijs in op $25' in plaats van 'voeg $ 5 toe aan de productprijs'. U kunt het eerste bericht een willekeurig aantal keren veilig verwerken en het resultaat is hetzelfde. Dat geldt niet voor het tweede bericht. Maar zelfs in het eerste geval wilt u de eerste gebeurtenis misschien niet verwerken, omdat het systeem ook een nieuwere prijswijzigingsgebeurtenis heeft verzonden en u de nieuwe prijs overschrijft.

Een ander voorbeeld kan een bestelling-voltooid evenement zijn dat is gepropageerd naar meerdere abonnees. De app moet ervoor zorgen dat ordergegevens slechts eenmaal in andere systemen worden bijgewerkt, zelfs als er dubbele berichtgebeurtenissen zijn voor dezelfde order voltooide gebeurtenis.

Het is handig om een soort identiteit per gebeurtenis te hebben, zodat u logica kunt maken die afdwingt dat elke gebeurtenis slechts eenmaal per ontvanger wordt verwerkt.

Sommige berichtverwerking is van nature idempotent. Als een systeem bijvoorbeeld afbeeldingsminiaturen genereert, maakt het mogelijk niet uit hoe vaak het bericht over de gegenereerde miniatuur wordt verwerkt; het resultaat is dat de miniaturen worden gegenereerd en dat ze elke keer hetzelfde zijn. Aan de andere kant zijn bewerkingen zoals het gebruik van een betaalgateway om een creditcard te belasten mogelijk helemaal niet idempotent. In dergelijke gevallen moet u ervoor zorgen dat het verwerken van een bericht meerdere keren het effect heeft dat u verwacht.

Aanvullende bronnen

Gebeurtenisberichten van integratie ontdubbelen

U kunt ervoor zorgen dat berichtengebeurtenissen slechts één keer per abonnee op verschillende niveaus worden verzonden en verwerkt. Een manier is om een ontdubbelingsfunctie te gebruiken die wordt aangeboden door de berichteninfrastructuur die u gebruikt. Een andere is het implementeren van aangepaste logica in uw doelmicroservice. Validaties op zowel het transportniveau als het toepassingsniveau zijn de beste optie.

Berichtgebeurtenissen dedupliceren op EventHandler-niveau

Een manier om ervoor te zorgen dat een gebeurtenis slechts eenmaal door een ontvanger wordt verwerkt, is door bepaalde logica te implementeren bij het verwerken van de berichtgebeurtenissen in gebeurtenis-handlers. Dit is bijvoorbeeld de methode die wordt gebruikt in de toepassing eShopOnContainers, zoals u kunt zien in de broncode van de klasse UserCheckoutAcceptedIntegrationEventHandler wanneer deze een UserCheckoutAcceptedIntegrationEvent integratiegebeurtenis ontvangt. (In dit geval wordt het CreateOrderCommand verpakt met een IdentifiedCommand, met behulp van de eventMsg.RequestId als id, voordat deze naar de opdrachthandler wordt verzonden).

Berichten ontdubbelen bij het gebruik van RabbitMQ

Wanneer er onregelmatige netwerkfouten optreden, kunnen berichten worden gedupliceerd en moet de ontvanger van het bericht gereed zijn om deze gedupliceerde berichten af te handelen. Indien mogelijk moeten ontvangers berichten op een idempotente manier verwerken, wat beter is dan expliciet omgaan met ontdubbeling.

Volgens de RabbitMQ-documentatie, "Als een bericht aan een consument wordt bezorgd en vervolgens opnieuw wordt ingeleverd omdat het niet is bevestigd voordat de verbinding van de consument werd verbroken, zal RabbitMQ de `redelivery` vlag zetten wanneer het opnieuw wordt afgeleverd aan dezelfde of een andere consument."

Als de vlag 'opnieuw verzenden' is ingesteld, moet de ontvanger daar rekening mee houden, omdat het bericht mogelijk al is verwerkt. Maar dat is niet gegarandeerd; het bericht mogelijk nooit de ontvanger heeft bereikt nadat het de berichtbroker heeft verlaten, mogelijk vanwege netwerkproblemen. Aan de andere kant, als de vlag 'opnieuw verzenden' niet is ingesteld, is het gegarandeerd dat het bericht niet meer dan één keer is verzonden. Daarom moet de ontvanger berichten ontdubbelen of berichten op een idempotente manier verwerken, alleen als de vlag 'opnieuw bezorgd' in het bericht gezet is.

Aanvullende bronnen

VorigeVolgende

  • Last updated on