versión:

Getting Started to Quarkus Messaging with AMQP 1.0

This guide demonstrates how your Quarkus application can utilize Quarkus Messaging to interact with AMQP 1.0.

If you want to use RabbitMQ, you should use the Quarkus Messaging RabbitMQ extension. Alternatively, if you want to use RabbitMQ with AMQP 1.0 you need to enable the AMQP 1.0 plugin in the RabbitMQ broker; check the connecting to RabbitMQ documentation.

Requisitos previos

To complete this guide, you need:

Roughly 15 minutes
An IDE
JDK 17+ installed with JAVA_HOME configured appropriately
Apache Maven 3.9.6
Docker and Docker Compose or Podman, and Docker Compose
Optionally the Quarkus CLI if you want to use it
Optionally Mandrel or GraalVM installed and configured appropriately if you want to build a native executable (or Docker if you use a native container build)

Arquitectura

In this guide, we are going to develop two applications communicating with an AMQP broker. We will use Artemis, but you can use any AMQP 1.0 broker. The first application sends a quote request to an AMQP queue and consumes messages from the quote queue. The second application receives the quote request and sends a quote back.

The first application, the producer, will let the user request some quotes over an HTTP endpoint. For each quote request, a random identifier is generated and returned to the user, to put the quote request on pending. At the same time the generated request id is sent over the quote-requests queue.

La segunda aplicación, la processor, a su vez, leerá de la cola quote-requests poner un precio aleatorio a la cotización, y lo enviará a una cola llamada quotes.

Lastly, the producer will read the quotes and send them to the browser using server-sent events. The user will therefore see the quote price updated from pending to the received price in real-time.

Solución

We recommend that you follow the instructions in the next sections and create applications step by step. However, you can go right to the completed example.

Clone el repositorio Git: git clone https://github.com/quarkusio/quarkus-quickstarts.git o descargue un archivo.

The solution is located in the amqp-quickstart directory.

Creación del proyecto Maven

En primer lugar, tenemos que crear dos proyectos: el productor y el procesador.

Para crear el proyecto productor, en un terminal ejecute:

CLI

quarkus create app org.acme:amqp-quickstart-producer \
    --extension='rest-jackson,messaging-amqp' \
    --no-code

To create a Gradle project, add the --gradle or --gradle-kotlin-dsl option.

For more information about how to install and use the Quarkus CLI, see the Quarkus CLI guide.

Maven

mvn io.quarkus.platform:quarkus-maven-plugin:3.9.4:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=amqp-quickstart-producer \
    -Dextensions='rest-jackson,messaging-amqp' \
    -DnoCode

To create a Gradle project, add the -DbuildTool=gradle or -DbuildTool=gradle-kotlin-dsl option.

For Windows users:

If using cmd, (don’t use backward slash \ and put everything on the same line)
If using Powershell, wrap -D parameters in double quotes e.g. "-DprojectArtifactId=amqp-quickstart-producer"

Este comando crea la estructura del proyecto y selecciona las dos extensiones de Quarkus que vamos a utilizar:

Quarkus REST (formerly RESTEasy Reactive) and its Jackson support to handle JSON payloads
El conector AMQP de mensajería reactiva

Para crear el proyecto del procesador, desde el mismo directorio, ejecute:

CLI

quarkus create app org.acme:amqp-quickstart-processor \
    --extension='messaging-amqp' \
    --no-code

To create a Gradle project, add the --gradle or --gradle-kotlin-dsl option.

For more information about how to install and use the Quarkus CLI, see the Quarkus CLI guide.

Maven

mvn io.quarkus.platform:quarkus-maven-plugin:3.9.4:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=amqp-quickstart-processor \
    -Dextensions='messaging-amqp' \
    -DnoCode

To create a Gradle project, add the -DbuildTool=gradle or -DbuildTool=gradle-kotlin-dsl option.

For Windows users:

If using cmd, (don’t use backward slash \ and put everything on the same line)
If using Powershell, wrap -D parameters in double quotes e.g. "-DprojectArtifactId=amqp-quickstart-processor"

En ese momento deberías tener la siguiente estructura:

.
├── amqp-quickstart-processor
│  ├── README.md
│  ├── mvnw
│  ├── mvnw.cmd
│  ├── pom.xml
│  └── src
│     └── main
│        ├── docker
│        ├── java
│        └── resources
│           └── application.properties
└── amqp-quickstart-producer
   ├── README.md
   ├── mvnw
   ├── mvnw.cmd
   ├── pom.xml
   └── src
      └── main
         ├── docker
         ├── java
         └── resources
            └── application.properties

Abra los dos proyectos en su IDE favorito.

El objeto Cotización

The Quote class will be used in both producer and processor projects. For the sake of simplicity we will duplicate the class. In both projects, create the src/main/java/org/acme/amqp/model/Quote.java file, with the following content:

package org.acme.amqp.model;

import io.quarkus.runtime.annotations.RegisterForReflection;

@RegisterForReflection
public class Quote {

    public String id;
    public int price;

    /**
    * Default constructor required for Jackson serializer
    */
    public Quote() { }

    public Quote(String id, int price) {
        this.id = id;
        this.price = price;
    }

    @Override
    public String toString() {
        return "Quote{" +
                "id='" + id + '\'' +
                ", price=" + price +
                '}';
    }
}

JSON representation of Quote objects will be used in messages sent to the AMQP queues and also in the server-sent events sent to browser clients.

Quarkus tiene capacidades incorporadas para tratar con mensajes JSON AMQP.

@RegisterForReflection

The @RegisterForReflection annotation instructs Quarkus to include the class (including fields and methods) when building the native executable. This will be useful later when we run the applications as native executables inside containers. Without, the native compilation would remove the fields and methods during the dead-code elimination phase.

Envío de solicitud de presupuesto

Inside the producer project locate the generated src/main/java/org/acme/amqp/producer/QuotesResource.java file, and update the content to be:

package org.acme.amqp.producer;

import java.util.UUID;

import jakarta.ws.rs.GET;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;

import org.acme.amqp.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

import io.smallrye.mutiny.Multi;

@Path("/quotes")
public class QuotesResource {

    @Channel("quote-requests") Emitter<String> quoteRequestEmitter; (1)

    /**
     * Endpoint to generate a new quote request id and send it to "quote-requests" AMQP queue using the emitter.
     */
    @POST
    @Path("/request")
    @Produces(MediaType.TEXT_PLAIN)
    public String createRequest() {
        UUID uuid = UUID.randomUUID();
        quoteRequestEmitter.send(uuid.toString()); (2)
        return uuid.toString();
    }
}

1	Inyectar una Mensajería Reactiva `Emitter` para enviar mensajes al canal `quote-requests`.
2	En una solicitud de publicación, genera un UUID aleatorio y lo envía a la cola AMQP utilizando el emisor.

The quote-requests channel is going to be managed as a AMQP queue, as that’s the only connector on the classpath. If not indicated otherwise, like in this example, Quarkus uses the channel name as AMQP queue name. So, in this example, the application sends messages to the quote-requests queue.

Si tiene varios conectores, deberá indicar qué conector desea utilizar en la configuración de la aplicación.

Tramitación de solicitudes de quota

Now let’s consume the quote request and give out a price. Inside the processor project, locate the src/main/java/org/acme/amqp/processor/QuoteProcessor.java file and add the following:

package org.acme.amqp.processor;

import java.util.Random;

import jakarta.enterprise.context.ApplicationScoped;

import org.acme.amqp.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Outgoing;

import io.smallrye.reactive.messaging.annotations.Blocking;

/**
 * A bean consuming data from the "request" AMQP queue and giving out a random quote.
 * The result is pushed to the "quotes" AMQP queue.
 */
@ApplicationScoped
public class QuoteProcessor {

    private Random random = new Random();

    @Incoming("requests")       (1)
    @Outgoing("quotes")         (2)
    @Blocking                   (3)
    public Quote process(String quoteRequest) throws InterruptedException {
        // simulate some hard-working task
        Thread.sleep(200);
        return new Quote(quoteRequest, random.nextInt(100));
    }
}

1	Indica que el método consume los elementos del canal `requests`
2	Indica que los objetos devueltos por el método se envían al canal `quotes`
3	Indica que el procesamiento está bloqueando y no puede ser ejecutado en el hilo de la llamada.

El método process es llamado por cada mensaje AMQP de la cola quote-requests, y enviará un objeto Quote a la cola quotes.

Because we want to consume messages from the quotes-requests queue into the requests channel, we need to configure this association. Open the src/main/resources/application.properties file and add:

mp.messaging.incoming.requests.address=quote-requests

The configuration properties are structured as follows:

mp.messaging.[outgoing|incoming].{channel-name}.property=value

En nuestro caso, queremos configurar el atributo address para indicar el nombre de la cola.

Recepción de presupuestos

Back to our producer project. Let’s modify the QuotesResource to consume quotes, bind it to an HTTP endpoint to send events to clients:

import io.smallrye.mutiny.Multi;
//...

@Channel("quotes") Multi<Quote> quotes;     (1)

/**
 * Endpoint retrieving the "quotes" queue and sending the items to a server sent event.
 */
@GET
@Produces(MediaType.SERVER_SENT_EVENTS) (2)
public Multi<Quote> stream() {
    return quotes; (3)
}

1	Inyecta el canal `quotes` utilizando el calificador `@Channel`
2	Indica que el contenido se envía utilizando `Server Sent Events`
3	Devuelve el flujo (Reactive Stream)

La página HTML

El toque final, la página HTML que lee los precios convertidos usando SSE.

Cree dentro del proyecto producer el archivo src/main/resources/META-INF/resources/quotes.html, con el siguiente contenido:

<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Quotes</title>

    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly.min.css">
    <link rel="stylesheet" type="text/css"
          href="https://cdnjs.cloudflare.com/ajax/libs/patternfly/3.24.0/css/patternfly-additions.min.css">
</head>
<body>
<div class="container">
    <div class="card">
        <div class="card-body">
            <h2 class="card-title">Quotes</h2>
            <button class="btn btn-info" id="request-quote">Request Quote</button>
            <div class="quotes"></div>
        </div>
    </div>
</div>
</body>
<script src="https://code.jquery.com/jquery-3.6.0.min.js"></script>
<script>
    $("#request-quote").click((event) => {
        fetch("/quotes/request", {method: "POST"})
        .then(res => res.text())
        .then(qid => {
            var row = $(`<h4 class='col-md-12' id='${qid}'>Quote # <i>${qid}</i> | <strong>Pending</strong></h4>`);
            $(".quotes").append(row);
        });
    });
    var source = new EventSource("/quotes");
    source.onmessage = (event) => {
      var json = JSON.parse(event.data);
      $(`#${json.id}`).html(function(index, html) {
        return html.replace("Pending", `\$\xA0${json.price}`);
      });
    };
</script>
</html>

Nothing spectacular here. On each received quote, it updates the page.

Ponerlo en marcha

Sólo tienes que ejecutar ambas aplicaciones utilizando:

> mvn -f amqp-quickstart-producer quarkus:dev

Y, en una terminal separada:

> mvn -f amqp-quickstart-processor quarkus:dev

Quarkus starts a AMQP broker automatically, configures the application and shares the broker instance between different applications. See Dev Services for AMQP for more details.

Abra http://localhost:8080/quotes.html en su navegador y solicite algunos presupuestos haciendo clic en el botón.

Ejecución en modo JVM o nativo

When not running in dev or test mode, you will need to start your AMQP broker. You can follow the instructions from the Apache ActiveMQ Artemis website or create a docker-compose.yaml file with the following content:

version: '2'

services:

  artemis:
    image: quay.io/artemiscloud/activemq-artemis-broker:1.0.25
    ports:
      - "8161:8161"
      - "61616:61616"
      - "5672:5672"
    environment:
      AMQ_USER: quarkus
      AMQ_PASSWORD: quarkus
    networks:
      - amqp-quickstart-network

  producer:
    image: quarkus-quickstarts/amqp-quickstart-producer:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: amqp-quickstart-producer
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      AMQP_HOST: artemis
      AMQP_PORT: 5672
    ports:
      - "8080:8080"
    networks:
      - amqp-quickstart-network

  processor:
    image: quarkus-quickstarts/amqp-quickstart-processor:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: amqp-quickstart-processor
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      AMQP_HOST: artemis
      AMQP_PORT: 5672
    networks:
      - amqp-quickstart-network

networks:
  amqp-quickstart-network:
    name: amqp-quickstart

Note how the AMQP broker location is configured. The amqp.host and amqp.port (AMQP_HOST and AMQP_PORT environment variables) properties configure location.

En primer lugar, asegúrese de que ha detenido las aplicaciones, y construya ambas aplicaciones en modo JVM con:

> mvn -f amqp-quickstart-producer clean package
> mvn -f amqp-quickstart-processor clean package

Once packaged, run docker compose up --build. The UI is exposed on http://localhost:8080/quotes.html

Para ejecutar tus aplicaciones como nativas, primero tenemos que construir los ejecutables nativos:

> mvn -f amqp-quickstart-producer package -Dnative  -Dquarkus.native.container-build=true
> mvn -f amqp-quickstart-processor package -Dnative -Dquarkus.native.container-build=true

The -Dquarkus.native.container-build=true instructs Quarkus to build Linux 64bits native executables, who can run inside containers. Then, run the system using:

> export QUARKUS_MODE=native
> docker compose up --build

Como antes, la interfaz de usuario está expuesta en http://localhost:8080/quotes.html

Ir más allá

This guide has shown how you can interact with AMQP 1.0 using Quarkus. It utilizes SmallRye Reactive Messaging to build data streaming applications.

If you did the Kafka quickstart, you have realized that it’s the same code. The only difference is the connector configuration and the JSON mapping.

Requisitos previos
Arquitectura
Solución
Creación del proyecto Maven
El objeto Cotización
Envío de solicitud de presupuesto
Tramitación de solicitudes de quota
Recepción de presupuestos
La página HTML
Ponerlo en marcha
Ejecución en modo JVM o nativo
Ir más allá