streamalism.org

1. Introduksjon

I denne opplæringen vil vi utforske begrepet fanout og emneutveksling med Spring AMQP og RabbitMQ.

På høyt nivå, fanout utveksling vil kringkaste den samme meldingen til alle bundne køer, samtidig som emneutveksling bruk en rutingnøkkel for overføring av meldinger til en bestemt bundet kø eller kø.

Førre lesing av Messaging With Spring AMQP anbefales for denne opplæringen.

2. Sette opp en Fanout Exchange

La oss sette opp en fanout-utveksling med to køer bundet til den. Når vi sender en melding til denne sentralen, mottar begge køene meldingen. Vår fanout-utveksling ignorerer enhver rutingnøkkel som følger med meldingen.

Våren AMQP lar oss samle alle erklæringer om køer, bytter og bindinger i en Erklærbare gjenstand:

@Bean public Declarables fanoutBindings () {Queue fanoutQueue1 = new Queue ("fanout.queue1", false); Queue fanoutQueue2 = ny kø ("fanout.queue2", false); FanoutExchange fanoutExchange = ny FanoutExchange ("fanout.exchange"); returner nye Declarables (fanoutQueue1, fanoutQueue2, fanoutExchange, bind (fanoutQueue1) .to (fanoutExchange), BindingBuilder.bind (fanoutQueue2) .to (fanoutExchange)); }

3. Sette opp en emneutveksling

Nå vil vi også sette opp en emneutveksling med to køer, hver med et annet bindingsmønster:

@Bean public Declarables topicBindings () {Kø topicQueue1 = ny kø (topicQueue1Name, false); Kø topicQueue2 = ny kø (topicQueue2Name, false); TopicExchange topicExchange = ny TopicExchange (topicExchangeName); returner nye Declarables (topicQueue1, topicQueue2, topicExchange, BindingBuilder .bind (topicQueue1) .to (topicExchange) .with ("*. important. *"), BindingBuilder .bind (topicQueue2) .to (topicExchange) .with ("#. feil")); }

En emneutveksling lar oss binde køer til den med forskjellige nøkkelmønstre. Dette er veldig fleksibelt og lar oss binde flere køer med samme mønster eller til og med flere mønstre til samme kø.

Når meldingens rutingnøkkel samsvarer med mønsteret, blir den plassert i køen. Hvis en kø har flere bindinger som samsvarer med meldingens rutingnøkkel, plasseres bare en kopi av meldingen i køen.

Våre bindingsmønstre kan bruke en stjerne (“*”) for å matche et ord i en bestemt posisjon eller et pundtegn (“#”) for å matche null eller flere ord.

Så, vår topicQue1 vil motta meldinger som har rutetaster som har et tre-ords mønster med mellomordet som "viktig" - for eksempel: “User.important.error” eller “Blog.important.notification”.

Og vår topicQue2 vil motta meldinger som har rutetaster som ender på ordfeilen; samsvarende eksempler er "feil", “User.important.error” eller “Blog.post.save.error”.

4. Sette opp en produsent

Vi bruker convertAndSend metoden for RabbitTemplate for å sende eksempler på meldinger:

 Strengmelding = "nyttelast sendes"; return args -> {rabbitTemplate.convertAndSend (FANOUT_EXCHANGE_NAME, "", "fanout" + melding); rabbitTemplate.convertAndSend (TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_WARN, "emne viktig advarsel" + melding); rabbitTemplate.convertAndSend (TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_ERROR, "emne viktig feil" + melding); };

De RabbitTemplate gir mange overbelastede convertAndSend () metoder for forskjellige utvekslingstyper.

Når vi sender en melding til en fanout-sentral, ignoreres rutingnøkkelen, og meldingen sendes til alle avgrensede køer.

Når vi sender en melding til emneutvekslingen, må vi sende en rutingnøkkel. Basert på denne rutingsnøkkelen vil meldingen bli levert til bestemte køer.

5. Konfigurere forbrukere

Til slutt, la oss sette opp fire forbrukere - en for hver kø - for å hente de produserte meldingene:

 @RabbitListener (køer = {FANOUT_QUEUE_1_NAME}) offentlig ugyldig mottaMessageFromFanout1 (strengmelding) {System.out.println ("Mottatt fanout 1 melding:" + melding); } @RabbitListener (køer = {FANOUT_QUEUE_2_NAME}) offentlig ugyldig mottaMessageFromFanout2 (strengmelding) {System.out.println ("Mottatt fanout 2-melding:" + melding); } @RabbitListener (køer = {TOPIC_QUEUE_1_NAME}) offentlig ugyldig motta MessageFromTopic1 (strengmelding) {System.out.println ("Mottatt emne 1 (" + BINDING_PATTERN_IMPORTANT + ") melding:" + melding); } @RabbitListener (køer = {TOPIC_QUEUE_2_NAME}) offentlig ugyldig motta MeldingFromTopic2 (strengmelding) {System.out.println ("Mottatt emne 2 (" + BINDING_PATTERN_ERROR + ") melding:" + melding); }

Vi konfigurerer forbrukerne ved hjelp av @RabbitListener kommentar. Det eneste argumentet som sendes her er køenes navn. Forbrukerne er ikke klar over utveksling eller rutingnøkler her.

6. Kjører eksemplet

Eksempelprosjektet vårt er et Spring Boot-program, og det vil derfor initialisere applikasjonen sammen med en forbindelse til RabbitMQ og sette opp alle køer, utvekslinger og bindinger.

Som standard forventer applikasjonen vår en RabbitMQ-forekomst som kjører på localhost på port 5672. Vi kan endre dette og andre standardinnstillinger i application.yaml.

Prosjektet vårt avslører HTTP-sluttpunkt på URI - /kringkaste - som godtar innlegg med en melding i forespørselsorganet.

Når vi sender en forespørsel til denne URI med body “Test”, bør vi se noe som ligner på dette i utdataene:

Mottatt fanout 1 melding: fanout nyttelast sendes Mottatt emne 1 (*. Viktig. *) Melding: emne viktig advarsel nyttelast sendes Mottatt emne 2 (# .feil) melding: emne viktig feil nyttelast kringkastes Mottatt fanout 2 melding: fanout nyttelast sendes Mottatt emne 1 (* .viktig. *) melding: emne viktig feil nyttelast sendes

Rekkefølgen vi vil se disse meldingene i, er selvfølgelig ikke garantert.

7. Konklusjon

I denne raske opplæringen dekket vi fans og temautvekslinger med Spring AMQP og RabbitMQ.

Den komplette kildekoden og alle kodebitene for denne opplæringen er tilgjengelig på GitHub-depotet.