バージョン:

Apache PulsarによるQuarkusメッセージングの開始

このガイドでは、QuarkusアプリケーションがQuarkus Messagingを利用してApache Pulsarとやりとりする方法を示します。

前提条件

このガイドを完成させるには、以下が必要です:

約15分
IDE
JDK 17+がインストールされ、 JAVA_HOME が適切に設定されていること
Apache Maven 3.9.9
Docker と Docker Compose、または Podman 、および Docker Compose
使用したい場合は、 Quarkus CLI
ネイティブ実行可能ファイルをビルドしたい場合、MandrelまたはGraalVM（あるいはネイティブなコンテナビルドを使用する場合はDocker）をインストールし、適切に設定していること

アーキテクチャ

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.

最後に、プロデューサー は見積を読み取り、サーバーから送信されたイベントを使用してブラウザーに送信します。したがって、ユーザーには、見積価格が保留から受信した価格にリアルタイムで更新されていることがわかります。

ソリューション

次の章で紹介する手順に沿って、ステップを踏んでアプリを作成することをお勧めします。ただし、完成した例にそのまま進んでも構いません。

Gitレポジトリをクローンするか git clone https://github.com/quarkusio/quarkus-quickstarts.git 、アーカイブをダウンロードします。

The solution is located in the pulsar-quickstart directory.

Mavenプロジェクトの作成

まず、 producer と processor の2つのプロジェクトを作成する必要があります。

producer プロジェクトを作成するには、ターミナルで次のように実行します:

コマンドラインインタフェース

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

Gradleプロジェクトを作成するには、 --gradle または --gradle-kotlin-dsl オプションを追加します。

Quarkus CLIのインストールと使用方法の詳細については、 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

Gradleプロジェクトを作成するには、 -DbuildTool=gradle または -DbuildTool=gradle-kotlin-dsl オプションを追加します。

Windowsユーザーの場合:

cmdを使用する場合、（バックスラッシュ \ を使用せず、すべてを同じ行に書かないでください）。
Powershellを使用する場合は、 -D パラメータを二重引用符で囲んでください。例: "-DprojectArtifactId=pulsar-quickstart-producer"

このコマンドは、プロジェクト構造を作成し、使用する 2 つの Quarkus エクステンションを選択します。

Quarkus REST (formerly RESTEasy Reactive) and its Jackson support (to handle JSON) to serve the HTTP endpoint.
The Pulsar connector for Reactive Messaging

processor プロジェクトを作成するには、同じディレクトリから、次のように実行します:

コマンドラインインタフェース

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

Gradleプロジェクトを作成するには、 --gradle または --gradle-kotlin-dsl オプションを追加します。

Quarkus CLIのインストールと使用方法の詳細については、 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

Gradleプロジェクトを作成するには、 -DbuildTool=gradle または -DbuildTool=gradle-kotlin-dsl オプションを追加します。

Windowsユーザーの場合:

cmdを使用する場合、（バックスラッシュ \ を使用せず、すべてを同じ行に書かないでください）。
Powershellを使用する場合は、 -D パラメータを二重引用符で囲んでください。例: "-DprojectArtifactId=pulsar-quickstart-processor"

その時点で、次の構造になっているはずです。

.
├── 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

2つのプロジェクトをお好みのIDEで開きます。

Dev Services

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.

見積オブジェクト

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.

見積もり依頼の送信

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	Reactive Messaging `Emitter` を注入して、 `quote-requests` チャンネルにメッセージを送信します。
2	On a post request, generate a random UUID and send it to the Pulsar topic using the emitter.
3	同じ UUID をクライアントに返します。

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.

複数のコネクターがある場合は、アプリケーション設定で使用するコネクターを指定する必要があります。

見積依頼の処理

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	メソッドが `requests` チャネルからのアイテムを消費することを示します。
2	メソッドによって返されるオブジェクトが `quotes` チャネルに送信されることを示します。
3	処理が blocking であり、呼び出し元のスレッドでは実行できないことを示します。

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

この場合、着信側と発信側のコネクタ設定が1つずつあり、それぞれに明確な名前が付けられていることに注意してください。設定プロパティは以下のような構造になっています：

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

channel-name セグメントは、 @Incoming および @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.

見積の受け取り

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	`@Channel` 修飾子を使用して `quotes` のチャネルを注入します
2	`Server Sent Events` を使用してコンテンツが送信されたことを示します。
3	ストリーム (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.

HTML ページ

見積をリクエストし、SSE で取得した価格を表示する HTML ページの最終調整を行います。

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>

ここでは何も特別なことはありません。ユーザーがボタンをクリックすると、見積をリクエストするための HTTP リクエストが作成され、保留中の見積がリストに追加されます。SSE を介して受け取った見積もりごとに、リスト内の対応するアイテムが更新されます。

実行

両方のアプリケーションを実行する必要があります。1 つの端末で、次を実行します。

mvn -f pulsar-quickstart-producer quarkus:dev

別の端末で、次を実行します。

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.

http://localhost:8080/quotes.html をブラウザで開き、ボタンをクリックして見積を依頼してください。

JVM またはネイティブモードでの実行

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

最初に、次のコマンドを使用して両方のアプリケーションを JVM モードでビルドします。

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

パッケージ化したら、docker-compose up を実行します。

これは開発クラスターであり、本番では使用しないでください。

アプリケーションをネイティブ実行可能ファイルとしてビルドし、実行することもできます。まず、両方のアプリケーションをネイティブとしてコンパイルします。

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

次のコマンドでシステムを実行します。

export QUARKUS_MODE=native
docker-compose up --build

さらに詳しく

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.

前提条件
アーキテクチャ
ソリューション
Mavenプロジェクトの作成
見積オブジェクト
見積もり依頼の送信
見積依頼の処理
見積の受け取り
HTML ページ
実行
JVM またはネイティブモードでの実行
さらに詳しく

Apache PulsarによるQuarkusメッセージングの開始

前提条件

アーキテクチャ

ソリューション

Mavenプロジェクトの作成

見積オブジェクト

見積もり依頼の送信

見積依頼の処理

見積の受け取り

HTML ページ

実行

JVM またはネイティブモードでの実行

さらに詳しく

関連コンテンツ

On the same extensions

On the same topics