streamalism.org

1. Bemutatkozás

Ebben az oktatóanyagban feltárjuk a fogalmát rajongás és témacserék a Spring AMQP-vel és a RabbitMQ-val.

Magas szinten, rajongói cserék akarat ugyanazt az üzenetet sugározza az összes kötött sorba, miközben témacserék használjon útválasztó kulcsot a üzenetek továbbítása egy meghatározott kötött sorba vagy sorokba.

Ehhez az oktatóanyaghoz a Messaging with Spring AMQP előzetes elolvasása ajánlott.

2. Fanout Exchange beállítása

Állítsunk be egy rajongói központot, két sorral összekötve. Amikor üzenetet küldünk ennek a központnak, mindkét sor megkapja az üzenetet. A rajongói központunk figyelmen kívül hagy minden, az üzenethez tartozó útválasztási kulcsot.

A tavaszi AMQP lehetővé teszi számunkra, hogy összesítsük a várólisták, cserék és kötések összes deklarációját a Nyilatkozható tárgy:

@Bean public Declarables fanoutBindings () {Queue fanoutQueue1 = new Queue ("fanout.queue1", hamis); Queue fanoutQueue2 = új várólista ("fanout.queue2", hamis); FanoutExchange fanoutExchange = new FanoutExchange ("fanout.exchange"); adja vissza az új nyilatkozatokat (fanoutQueue1, fanoutQueue2, fanoutExchange, bind (fanoutQueue1) .to (fanoutExchange), BindingBuilder.bind (fanoutQueue2) .to (fanoutExchange)); }

3. Témacsere beállítása

Most létrehozunk egy témacserét is, két sorral, amelyek mindegyike eltérő kötési mintával rendelkezik:

@Bean public Declarables topicBindings () {Queue topicQueue1 = new Queue (topicQueue1Name, hamis); Queue topicQueue2 = új várólista (topicQueue2Name, hamis); TopicExchange topicExchange = új TopicExchange (topicExchangeName); adja vissza az új deklarálhatókat (topicQueue1, topicQueue2, topicExchange, BindingBuilder .bind (topicQueue1) .to (topicExchange) .with ("*. fontos. *"), BindingBuilder .bind (topicQueue2) .to (topicExchange) .with ("#. hiba")); }

A témacsere lehetővé teszi számunkra, hogy különböző kulcsmintákkal kössük hozzá a sorokat. Ez nagyon rugalmas, és lehetővé teszi számunkra, hogy több, ugyanazzal a mintával, vagy akár több mintát is azonos sorba kössünk.

Amikor az üzenet útválasztási kulcsa megegyezik a mintával, akkor a sorba kerül. Ha egy sornak több kötése van, amelyek megfelelnek az üzenet útválasztási kulcsának, akkor az üzenetnek csak egy példánya kerül a sorba.

Kötési mintáink csillaggal (“*”) illeszthetik egy adott helyzetben lévő szót, vagy font (“#”) nulla vagy több szó egyezéséhez.

Szóval, a mi topicQue1 üzeneteket fog kapni, amelyek háromszavas mintájú útválasztó kulcsokkal rendelkeznek, a középső szó pedig „fontos” - például: „User.important.error” vagy „Blog.fontos.értesítés”.

És, a mi topicQueue üzeneteket fog kapni, amelyeken a hibára végződő útválasztó kulcsok vannak; megfelelő példák "hiba", „User.important.error” vagy „Blog.post.save.error”.

4. Producer beállítása

Használjuk a convertAndSend módszere RabbitTemplate mintaüzeneteink elküldéséhez:

 Karakterlánc üzenet = "hasznos teher sugárzott"; return args -> {rabbitTemplate.convertAndSend (FANOUT_EXCHANGE_NAME, "", "fanout" + üzenet); rabbitTemplate.convertAndSend (TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_WARN, "téma fontos figyelmeztetés" + üzenet); rabbitTemplate.convertAndSend (TOPIC_EXCHANGE_NAME, ROUTING_KEY_USER_IMPORTANT_ERROR, "téma fontos hiba" + üzenet); };

A RabbitTemplate sok túlterheltet biztosít convertAndSend () módszerek a különböző cseretípusokhoz.

Amikor üzenetet küldünk egy fanout központnak, az útválasztási kulcs figyelmen kívül marad, és az üzenet minden kötött sorba továbbításra kerül.

Amikor üzenetet küldünk a témacserének, át kell adnunk egy útválasztási kulcsot. Ezen útválasztási kulcs alapján az üzenet meghatározott sorokba kerül.

5. A fogyasztók beállítása

Végül állítsunk be négy fogyasztót - egy-egy soronként - az előállított üzenetek felvételére:

 @RabbitListener (várólisták = {FANOUT_QUEUE_1_NAME}) public void ReceiveMessageFromFanout1 (String üzenet) {System.out.println ("Fogadott 1. rajongói üzenet:" + üzenet); } @RabbitListener (várólisták = {FANOUT_QUEUE_2_NAME}) public void fogadásiMessageFromFanout2 (String üzenet) {System.out.println ("Fogadott 2. rajongói üzenet:" + üzenet); } @RabbitListener (várólisták = {TOPIC_QUEUE_1_NAME}) public void ReceiveMessageFromTopic1 (karakterlánc-üzenet) {System.out.println ("Fogadott 1. téma (" + BINDING_PATTERN_IMPORTANT + ") üzenet:" + üzenet); } @RabbitListener (várólisták = {TOPIC_QUEUE_2_NAME}) public void ReceiveMessageFromTopic2 (String üzenet) {System.out.println ("Fogadott 2. téma (" + BINDING_PATTERN_ERROR + ") üzenet:" + üzenet); }

A fogyasztókat a @RabbitListener annotáció. Az egyetlen érv itt a sorok neve. A fogyasztók itt nincsenek tisztában a cserékkel vagy a kulcsok továbbításával.

6. A példa futtatása

A mintaprojektünk egy Spring Boot alkalmazás, így inicializálja az alkalmazást, a RabbitMQ-val való kapcsolattal együtt, és beállítja az összes várólistát, csomót és kötést.

Alapértelmezés szerint alkalmazásunk RabbitMQ példányt vár az 5672-es port localhost-ján. Ezt és más alapértelmezéseket a alkalmazás.yaml.

Projektünk HTTP végpontot tár fel az URI-n - /adás - amely a kérelem törzsében egy üzenettel fogadja a POST-okat.

Amikor egy kérést küldünk ennek az URI-nak a „Test” testtel, ehhez hasonlót kell látnunk a kimenetben:

Fogadott 1. fanout üzenet: fanout hasznos terhet sugároznak Fogadott 1. téma (* .fontos. *) Üzenet: a téma fontos figyelmeztetni a hasznos terhet sugárzott Fogadott 2. téma (# .hiba) üzenet: a témában fontos hiba hasznos adatot sugároznak Fogadott 2. fanout üzenet: fanout hasznos terhelés sugárzott Fogadott 1. téma (* .fontos. *) üzenet: a téma fontos hibája hasznos adatot sugároz

Ezeket az üzeneteket sorrendben természetesen nem garantáljuk.

7. Következtetés

Ebben a gyors bemutatóban a rajongók és a témák cseréjét ismertettük a Spring AMQP-vel és a RabbitMQ-val.

Az oktatóanyag teljes forráskódja és összes kódrészlete elérhető a GitHub adattárban.