Por Versão

Getting Started to Quarkus Messaging with Apache Pulsar

This guide demonstrates how your Quarkus application can utilize Quarkus Messaging to interact with Apache Pulsar.

Pré-requisitos

Para concluir este guia, você precisa:

Cerca de 15 minutos
Um IDE
JDK 17+ installed with JAVA_HOME configured appropriately
Apache Maven 3.9.9
Docker e Docker Compose ou Podman e Docker Compose
Opcionalmente, o Quarkus CLI se você quiser usá-lo
Opcionalmente, Mandrel ou GraalVM instalado e configurado apropriadamente se você quiser criar um executável nativo (ou Docker se você usar uma compilação de contêiner nativo)

Arquitetura

In this guide, we are going to develop two applications communicating with Pulsar. The first application sends a quote request to Pulsar and consumes Pulsar messages from the quote topic. 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 mark the quote request as pending. At the same time, the generated request id is sent over a Pulsar topic quote-requests.

The second application, the processor, will read from the quote-requests topic, put a random price to the quote, and send it to a Pulsar topic named quotes.

Por fim, o producer lerá as cotações e as enviará para o navegador usando Server-Sent Events. Assim, o usuário verá o preço da cotação atualizado de pending para o preço recebido em tempo real.

Solução

Recomendamos que siga as instruções nas próximas seções e crie aplicativos passo a passo. No entanto, você pode ir direto para o exemplo concluído.

Clone o repositório Git: git clone https://github.com/quarkusio/quarkus-quickstarts.git, ou baixe um arquivo.

The solution is located in the pulsar-quickstart directory.

Criando o projeto Maven

Em primeiro lugar, temos de criar dois projetos: o producer e o processor.

Para criar o projeto producer, em um terminal, execute:

CLI

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

Para criar um projeto Gradle, adicione a opção --gradle ou --gradle-kotlin-dsl.

Para obter mais informações sobre como instalar e usar a CLI do Quarkus, consulte o guia Quarkus CLI.

Maven

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

Para criar um projeto Gradle, adicione a opção '-DbuildTool=gradle' ou '-DbuildTool=gradle-kotlin-dsl'.

Para usuários do Windows:

Se estiver usando cmd, (não use barra invertida '\' e coloque tudo na mesma linha)
Se estiver usando o Powershell, envolva os parâmetros '-D' entre aspas duplas, por exemplo, '"-DprojectArtifactId=pulsar-quickstart-producer"'

Esse comando cria a estrutura do projeto e seleciona duas extensões do Quarkus que usaremos:

Quarkus REST (anteriormente RESTEasy Reactive) e seu suporte Jackson (para manipular JSON) para servir o endpoint HTTP.
The Pulsar connector for Reactive Messaging

Para criar o projeto do processor, a partir do mesmo diretório, execute:

CLI

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

Para criar um projeto Gradle, adicione a opção --gradle ou --gradle-kotlin-dsl.

Para obter mais informações sobre como instalar e usar a CLI do Quarkus, consulte o guia Quarkus CLI.

Maven

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

Para criar um projeto Gradle, adicione a opção '-DbuildTool=gradle' ou '-DbuildTool=gradle-kotlin-dsl'.

Para usuários do Windows:

Se estiver usando cmd, (não use barra invertida '\' e coloque tudo na mesma linha)
Se estiver usando o Powershell, envolva os parâmetros '-D' entre aspas duplas, por exemplo, '"-DprojectArtifactId=pulsar-quickstart-processor"'

Nesse ponto, você deve ter a seguinte estrutura:

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

Abra os dois projetos no seu IDE preferido.

Serviços de Desenvolvimento

No need to start a Pulsar broker when using the dev mode or for tests. Quarkus starts a broker for you automatically. See Dev Services for Pulsar for details.

O objeto Quote

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/pulsar/model/Quote.java file, with the following content:

package org.acme.pulsar.model;

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 Pulsar topic and also in the server-sent events sent to web browsers.

Quarkus has built-in capabilities to deal with JSON Pulsar messages. In a following section, we will create serializer/deserializer classes for Jackson.

Enviando pedido de cotação

Inside the producer project, create the src/main/java/org/acme/pulsar/producer/QuotesResource.java file and add the following content:

package org.acme.pulsar.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.pulsar.model.Quote;
import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;

@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" Pulsar topic 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(); (3)
    }
}

1	Injete um serviço de mensagens reativas `Emitter` para enviar mensagens para o canal `quote-requests`.
2	On a post request, generate a random UUID and send it to the Pulsar topic using the emitter.
3	Retorne o mesmo UUID para o cliente.

The quote-requests channel is going to be managed as a Pulsar topic, as that’s the only connector on the classpath. If not indicated otherwise, like in this example, Quarkus uses the channel name as topic name. So, in this example, the application writes into the quote-requests topic. Quarkus also configures the serializer automatically, because it finds that the Emitter produces String values.

Quando você tem vários conectores, você precisaria indicar qual conector deseja usar na configuração do aplicativo.

Processamento de pedidos de cotação

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

package org.acme.pulsar.processor;

import java.util.Random;

import jakarta.enterprise.context.ApplicationScoped;

import org.acme.pulsar.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 "quote-requests" Pulsar topic (mapped to "requests" channel) and giving out a random quote.
 * The result is pushed to the "quotes" Pulsar topic.
 */
@ApplicationScoped
public class QuotesProcessor {

    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 o método consome os itens do canal `requests`.
2	Indica que os objetos retornados pelo método são enviados para o canal `quotes`.
3	Indica que o processamento está a bloquear e não pode ser executado na thread do chamador.

For every Pulsar message from the quote-requests topic, Reactive Messaging calls the process method, and sends the returned Quote object to the quotes channel. In this case, we need to configure the channel in the application.properties file, to configures the requests and quotes channels:

%dev.quarkus.http.port=8081

# Configure the incoming `quote-requests` Pulsar topic
mp.messaging.incoming.requests.topic=quote-requests
mp.messaging.incoming.requests.subscriptionInitialPosition=Earliest

Observe que, neste caso, temos uma configuração de conector de entrada e uma de saída, cada uma com um nome distinto. As propriedades de configuração são estruturadas da seguinte forma:

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

O segmento channel-name deve corresponder ao valor definido na anotação @Incoming e @Outgoing:

quote-requests → Pulsar topic from which we read the quote requests
quotes → Pulsar topic in which we write the quotes

More details about this configuration is available on the https://pulsar.apache.org/docs/3.0.x/concepts-messaging/ section from the Pulsar documentation. These properties are configured with the prefix pulsar. An exhaustive list of configuration properties is available in Pulsar Reference Guide - Configuration.

mp.messaging.incoming.requests.subscriptionInitialPosition=Earliest instructs the application to start reading the topics from the first message on the topic, when there is no previously acked messages. In other words, it will also process messages sent before we start the processor application.

There is no need to set schemas. Quarkus detects them, and if none are found, generates them using appropriate schema types. Structured types like Quote bean use JSON schema.

Recebendo cotações

Back to our producer project. Let’s modify the QuotesResource to consume quotes from Pulsar and send them back to the client via Server-Sent Events:

import io.smallrye.mutiny.Multi;

...

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

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

1	Injecta o canal `quotes` utilizando o qualificador `@Channel`
2	Indica que o conteúdo é enviado utilizando `Server Sent Events`
3	Devolve o fluxo (Reactive Stream)

No need to configure anything, as Quarkus will automatically associate the quotes channel to the quotes Pulsar topic. It will also generate a deserializer for the Quote class.

Message Schemas in Pulsar

In this example we used JSON Schema with Pulsar messages. For more information on Pulsar Schemas, see Pulsar Reference Guide - Schema.

A página HTML

O toque final, a página HTML solicitando cotações e exibindo os preços obtidos por SSE.

Inside the pulsar-quickstart-producer project, create the src/main/resources/META-INF/resources/quotes.html file with the following content:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Prices</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").prepend(row);
        });
    });

    var source = new EventSource("/quotes");
    source.onmessage = (event) => {
      var json = JSON.parse(event.data);
      $(`#${json.id}`).html((index, html) => {
        return html.replace("Pending", `\$\xA0${json.price}`);
      });
    };
</script>
</html>

Nada espetacular aqui. Quando o usuário clica no botão, uma solicitação HTTP é feita para solicitar uma cotação, e uma cotação pendente é adicionada à lista. A cada cotação recebida por SSE, o item correspondente na lista é atualizado.

Executando a aplicação

Você só precisa executar ambas as aplicações. Em um terminal, execute:

mvn -f pulsar-quickstart-producer quarkus:dev

Em outro terminal, execute:

mvn -f pulsar-quickstart-processor quarkus:dev

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

Abra http://localhost:8080/quotes.html no seu browser e peça algumas cotações clicando no botão.

Execução em modo JVM ou nativo

When not running in dev or test mode, you will need to start your Pulsar broker. You can follow the instructions from the Run a standalone Pulsar cluster in Docker or create a docker-compose.yaml file with the following content:

version: '3.8'

services:

  pulsar:
    image: apachepulsar/pulsar:3.2.4
    command: [
      "sh", "-c",
      "bin/apply-config-from-env.py conf/standalone.conf && bin/pulsar standalone -nfw -nss"
    ]

    ports:
      - "6650:6650"
      - "8080:8080"
    tmpfs:
      - /pulsar/data
    healthcheck:
      test: curl --fail http://localhost:8080/admin/v2/clusters || exit 1
      interval: 10s
      timeout: 10s
      retries: 5
      start_period: 5s
    environment:
      PULSAR_PREFIX_advertisedListeners: internal:pulsar://localhost:6650,external:pulsar://pulsar:6650
      PULSAR_PREFIX_transactionCoordinatorEnabled: true
      PULSAR_PREFIX_systemTopicEnabled: true
    networks:
      - pulsar-quickstart-network

  producer:
    image: quarkus-quickstarts/pulsar-quickstart-producer:1.0-${QUARKUS_MODE:-jvm}
    depends_on:
      pulsar:
        condition: service_healthy
    build:
      context: pulsar-quickstart-producer
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    deploy:
      restart_policy:
        condition: on-failure
    environment:
      PULSAR_CLIENT_SERVICE_URL: pulsar://pulsar:6650
    ports:
      - "8082:8080"
    networks:
      - pulsar-quickstart-network

  processor:
    image: quarkus-quickstarts/pulsar-quickstart-processor:1.0-${QUARKUS_MODE:-jvm}
    depends_on:
      pulsar:
        condition: service_healthy
    build:
      context: pulsar-quickstart-processor
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    deploy:
      restart_policy:
        condition: on-failure
    environment:
      QUARKUS_HTTP_PORT: 8082
      PULSAR_CLIENT_SERVICE_URL: pulsar://pulsar:6650
    ports:
      - "8083:8080"
    networks:
      - pulsar-quickstart-network

networks:
  pulsar-quickstart-network:
    name: pulsar-quickstart

Certifique-se de construir primeiro ambas as aplicações no modo JVM com:

mvn -f pulsar-quickstart-producer package
mvn -f pulsar-quickstart-processor package

Uma vez empacotado, execute docker-compose up.

Este é um cluster de desenvolvimento, não use em produção.

Você também pode construir e executar nossas aplicações como executáveis nativos. Primeiro, compile ambas as aplicações como nativas:

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

Execute o sistema com:

export QUARKUS_MODE=native
docker-compose up --build

Indo mais longe

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

For the exhaustive list of features and configuration options, check the Reference guide for Apache Pulsar Extension.

In this guide we explore how to interact with Apache Pulsar using the Quarkus Messaging extensions. using Pulsar clients directly.

Pré-requisitos
Arquitetura
Solução
Criando o projeto Maven
O objeto Quote
Enviando pedido de cotação
Processamento de pedidos de cotação
Recebendo cotações
A página HTML
Executando a aplicação
Execução em modo JVM ou nativo
Indo mais longe

Getting Started to Quarkus Messaging with Apache Pulsar

Pré-requisitos

Arquitetura

Solução

Criando o projeto Maven

O objeto Quote

Enviando pedido de cotação

Processamento de pedidos de cotação

Recebendo cotações

A página HTML

Executando a aplicação

Execução em modo JVM ou nativo

Indo mais longe

Conteúdo Relacionado

Nas mesmas extensões

Nos mesmos tópicos