Getting Started to Quarkus Messaging with AMQP 1.0

本指南演示了 Quarkus 应用程序如何利用 Quarkus Messaging 与 AMQP 1.0 进行交互。

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

如果你想使用 RabbitMQ，你应使用 Quarkus Messaging RabbitMQ extension。或者，如果你想将 RabbitMQ 与 AMQP 1.0 一起使用，你需要在 RabbitMQ 代理中启用 AMQP 1.0 插件；查看 connecting to RabbitMQ文档。

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.

Prerequisites

Unresolved directive in amqp.adoc - include::{includes}/prerequisites.adoc[]

Architecture

在本指南中，我们将开发两个与 AMQP 代理通信的应用程序。我们将使用 Artemis，但你可以使用任何 AMQP 1.0 代理。第一个应用程序向 AMQP 队列发送 quote request，并使用 quote_队列中的消息。第二个应用程序接收 _quote request，然后发送回 quote。

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.

第一个应用程序 `producer`将让用户通过 HTTP 端点请求一些报价。对于每个报价请求，都会生成一个随机标识符并返回给用户，以将报价请求放在 _pending_上。同时，生成的请求 ID 会通过 `quote-requests`队列发送。

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.

第二个应用程序 `processor`反过来将从 `quote-requests`队列读取，向报价添加随机价格，并将其发送到名为 `quotes`的队列。

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

最后，producer 将读取报价并使用服务器端发送的事件将它们发送到浏览器。因此，用户将看到报价价格从 pending 实时更新为接收到的价格。

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.

Solution

我们建议您按照下一节中的说明一步步创建应用程序。但是，您可以直接转到已完成的示例。

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.

克隆 Git 存储库： git clone {quickstarts-clone-url}，或下载 {quickstarts-archive-url}[存档]。

Clone the Git repository: git clone {quickstarts-clone-url}, or download an {quickstarts-archive-url}[archive].

该解决方案位于 amqp-quickstart directory中。

The solution is located in the amqp-quickstart directory.

Creating the Maven Project

首先，我们需要创建两个项目：producer_和 _processor。

First, we need to create two projects: the producer and the processor.

要在终端中创建 _producer_项目，请运行：

To create the producer project, in a terminal run:

Unresolved directive in amqp.adoc - include::{includes}/devtools/create-app.adoc[]

此命令创建项目结构并选择我们将使用的两个 Quarkus 扩展：

This command creates the project structure and select the two Quarkus extensions we will be using:

Quarkus REST (formerly RESTEasy Reactive) and its Jackson support to handle JSON payloads
The Reactive Messaging AMQP connector

要从相同目录创建 _processor_项目，请运行：

To create the processor project, from the same directory, run:

Unresolved directive in amqp.adoc - include::{includes}/devtools/create-app.adoc[]

此时，您应该拥有以下结构：

At that point you should have the following structure:

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

在你的首选 IDE 中打开这两个项目。

Open the two projects in your favorite IDE.

The Quote object

该 `Quote`类将同时用于 `producer`和 `processor`项目。为简单起见，我们将复制该类。在这两个项目中，创建 `src/main/java/org/acme/amqp/model/Quote.java`文件，内容如下：

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 +
                '}';
    }
}

发送到 AMQP 队列的消息和发送到浏览器客户端的服务器端发送的事件都将使用 `Quote`对象的 JSON 表示形式。

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 具有处理 JSON AMQP 消息的内置功能。

Quarkus has built-in capabilities to deal with JSON AMQP messages.

@RegisterForReflection

@RegisterForReflection 注释指示 Quarkus 在创建本机可执行文件时保留类、其字段和方法。当我们稍后在容器中以本机可执行文件的形式运行应用程序时，这点很重要。如果没有此注释，本机编译过程将在死代码消除阶段丢弃字段和方法，这会导致运行时错误。有关 @RegisterForReflection 注释的更多详细信息，请参阅 native application tips 页面。

The @RegisterForReflection annotation instructs Quarkus to keep the class, its fields, and methods when creating a native executable. This is crucial when we later run our applications as native executables within containers. Without this annotation, the native compilation process would discard the fields and methods during the dead-code elimination phase, which would lead to runtime errors. More details about the @RegisterForReflection annotation can be found on the native application tips page.

Sending quote request

在 `producer`项目内部，找到生成的 `src/main/java/org/acme/amqp/producer/QuotesResource.java`文件，并将内容更新为：

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	Inject a Reactive Messaging `Emitter` to send messages to the `quote-requests` channel.
2	On a post request, generate a random UUID and send it to the AMQP queue using the emitter.

`quote-requests`通道将被管理为 AMQP 队列，因为那是类路径上的唯一连接器。如果没有另行指明，就像此示例中一样，Quarkus 会使用通道名称作为 AMQP 队列名称。因此，在本例中，该应用程序会向 `quote-requests`队列发送消息。

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.

当你有多个连接器时，你需要在应用程序配置中指明你要在其中使用哪个连接器。

When you have multiple connectors, you would need to indicate which connector you want to use in the application configuration.

Processing quote requests

现在，让我们使用该报价请求并给出价格。在 `processor`项目内部，找到 `src/main/java/org/acme/amqp/processor/QuoteProcessor.java`文件并添加以下内容：

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	Indicates that the method consumes the items from the `requests` channel
2	Indicates that the objects returned by the method are sent to the `quotes` channel
3	Indicates that the processing is blocking and cannot be run on the caller thread.

`process`方法是为来自 `quote-requests`队列的每条 AMQP 消息调用的，它会将 `Quote`对象发送到 `quotes`队列。

The process method is called for every AMQP message from the quote-requests queue, and will send a Quote object to the quotes queue.

因为我们希望使用 `requests`通道将消息从 `quotes-requests`队列中使用，我们需要配置此关联。打开 `src/main/resources/application.properties`文件并添加：

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

在我们的示例中，我们要配置 `address`属性以指明队列名称。

In our case, we want to configure the address attribute to indicate the name of the queue.

Receiving quotes

回到我们的 producer 项目。让我们修改 QuotesResource 以使用报价、将其绑定到一个 HTTP 端点以向客户端发送事件：

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	Injects the `quotes` channel using the `@Channel` qualifier
2	Indicates that the content is sent using `Server Sent Events`
3	Returns the stream (Reactive Stream)

The HTML page

最后一步，使用 SSE 读取已转换价格的 HTML 页面。

Final touch, the HTML page reading the converted prices using SSE.

在 producer 项目内创建 src/main/resources/META-INF/resources/quotes.html 文件，并包含以下内容：

Create inside the producer project src/main/resources/META-INF/resources/quotes.html file, with the following content:

<!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.

Get it running

你只需要使用以下命令运行这两个应用程序：

You just need to run both applications using:

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

在另一个终端中：

And, in a separate terminal:

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

Quarkus 自动启动 AMQP 代理，为应用程序进行配置，并在不同应用程序之间共享代理实例，有关更多详细信息，请见 Dev Services for AMQP。

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.

在浏览器中打开 http://localhost:8080/quotes.html，然后通过单击按钮请求一些报价。

Open http://localhost:8080/quotes.html in your browser and request some quotes by clicking the button.

Running in JVM or Native mode

在非开发或测试模式下运行时，你需要启动 AMQP 代理。你可以遵循 Apache ActiveMQ Artemis website中的说明或创建一个 `docker-compose.yaml`文件，其内容如下：

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

注意 AMQP 代理位置的配置方式。amqp.host`和 `amqp.port(`AMQP_HOST`和 `AMQP_PORT`环境变量) 属性配置位置。

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

首先，确保已停止应用程序，并使用以下命令以 JVM 模式构建这两个应用程序：

First, make sure you stopped the applications, and build both applications in JVM mode with:

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

打包完成后，运行 docker compose up --build。用户界面在 [role="bare"][role="bare"]http://localhost:8080/quotes.html 上公开

Once packaged, run docker compose up --build. The UI is exposed on [role="bare"]http://localhost:8080/quotes.html

要以本机方式运行应用程序，我们首先需要构建本机可执行文件：

To run your applications as native, first we need to build the native executables:

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

-Dquarkus.native.container-build=true 指示 Quarkus 构建可在容器内运行的 Linux 64 位本机可执行文件。然后，使用以下命令运行系统：

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

和之前一样，用户界面在 [role="bare"][role="bare"]http://localhost:8080/quotes.html 上公开

As before, the UI is exposed on [role="bare"]http://localhost:8080/quotes.html

Going further

本指南展示了 Quarkus 如何与 AMQP 1.0 交互。它利用 SmallRye Reactive Messaging构建数据流应用程序。

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

如果你已经完成 Kafka 快速入门，你就会发现它是同样的代码。唯一的差异是连接器配置和 JSON 映射。

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.