The English version of quarkus.io is the official project site. Translated sites are community supported on a best-effort basis.
返回指南目录

Getting Started to Quarkus Messaging with RabbitMQ

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

这项技术被认为是preview。

在 preview(预览) 中，不保证向后兼容和在生态系统中的存在。具体的改进可能需要改变配置或API，并且正在计划变得 稳定 。欢迎在我们的 邮件列表 中提供反馈，或在我们的 GitHub问题列表 中提出问题。

For a full list of possible statuses, check our FAQ entry.

先决条件

完成这个指南，你需要:

  • 大概15分钟

  • 编辑器

  • JDK 17+ installed with JAVA_HOME configured appropriately

  • Apache Maven 3.9.9

  • Docker and Docker Compose or Podman, and Docker Compose

  • 如果你愿意的话，还可以选择使用Quarkus CLI

  • 如果你想构建原生可执行程序，可以选择安装Mandrel或者GraalVM，并正确配置(或者使用Docker在容器中进行构建)

应用结构

In this guide, we are going to develop two applications communicating with a RabbitMQ broker. The first application sends a quote request to the RabbitMQ quote requests exchange and consumes messages from the quote queue. The second application receives the quote request and sends a quote back.

Architecture

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 to the quote-requests exchange.

Producer App UI

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 an exchange named quotes.

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

解决方案

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

克隆 Git 仓库: git clone https://github.com/quarkusio/quarkus-quickstarts.git ，或下载一个 存档 。

The solution is located in the rabbitmq-quickstart directory.

创建Maven项目

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

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

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

创建Grade项目，请添加 --gradle 或者 --gradle-kotlin-dsl 参数。

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.16.3:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=rabbitmq-quickstart-producer \
    -Dextensions='messaging-rabbitmq,rest-jackson' \
    -DnoCode

创建Grade项目，请添加 -DbuildTool=gradle 或者 -DbuildTool=gradle-kotlin-dsl 参数。

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=rabbitmq-quickstart-producer"

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

  1. The Reactive Messaging RabbitMQ connector

  2. Quarkus REST (formerly RESTEasy Reactive) and its Jackson support to handle JSON payloads

If you already have your Quarkus project configured, you can add the messaging-rabbitmq extension to your project by running the following command in your project base directory:

CLI
quarkus extension add messaging-rabbitmq
Maven
./mvnw quarkus:add-extension -Dextensions='messaging-rabbitmq'
Gradle
./gradlew addExtension --extensions='messaging-rabbitmq'

这将在你的 pom.xml 中添加以下内容：

pom.xml
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-messaging-rabbitmq</artifactId>
</dependency>
build.gradle
implementation("io.quarkus:quarkus-messaging-rabbitmq")

要创建 processor 项目，请在同一目录下运行：

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

创建Grade项目，请添加 --gradle 或者 --gradle-kotlin-dsl 参数。

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.16.3:create \
    -DprojectGroupId=org.acme \
    -DprojectArtifactId=rabbitmq-quickstart-processor \
    -Dextensions='messaging-rabbitmq' \
    -DnoCode

创建Grade项目，请添加 -DbuildTool=gradle 或者 -DbuildTool=gradle-kotlin-dsl 参数。

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=rabbitmq-quickstart-processor"

此时，你应该有如下的结构：

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

在你喜欢的IDE中打开这两个项目。

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

package org.acme.rabbitmq.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 RabbitMQ queues and also in the server-sent events sent to browser clients.

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

@RegisterForReflection

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.

发送报价请求

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

package org.acme.rabbitmq.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.rabbitmq.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" channel (which
     * maps to the "quote-requests" RabbitMQ exchange) 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 注入一个响应式消息 Emitter ，来向 quote-requests 通道发送消息。
2 On a post request, generate a random UUID and send it to the RabbitMQ queue using the emitter.

This channel is mapped to a RabbitMQ exchange using the configuration we will add to the application.properties file. Open the src/main/resource/application.properties file and add:

# Configure the outgoing RabbitMQ exchange `quote-requests`
mp.messaging.outgoing.quote-requests.connector=smallrye-rabbitmq
mp.messaging.outgoing.quote-requests.exchange.name=quote-requests

All we need to specify is the smallrye-rabbitmq connector. By default, reactive messaging maps the channel name quote-requests to the same RabbitMQ exchange name.

处理报价请求

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

package org.acme.rabbitmq.processor;

import java.util.Random;

import jakarta.enterprise.context.ApplicationScoped;

import org.acme.rabbitmq.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" RabbitMQ queue and giving out a random quote.
 * The result is pushed to the "quotes" RabbitMQ exchange.
 */
@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(1000);
        return new Quote(quoteRequest, random.nextInt(100));
    }
}
1 表示该方法消费 requests 通道中的项
2 表示方法返回的对象被发送给 quotes 通道
3 表示该处理是 blocking ，不能在调用者线程上运行。

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

As with the previous example we need to configure the connectors in the application.properties file. Open the src/main/resources/application.properties file and add:

# Configure the incoming RabbitMQ queue `quote-requests`
mp.messaging.incoming.requests.connector=smallrye-rabbitmq
mp.messaging.incoming.requests.queue.name=quote-requests
mp.messaging.incoming.requests.exchange.name=quote-requests

# Configure the outgoing RabbitMQ exchange `quotes`
mp.messaging.outgoing.quotes.connector=smallrye-rabbitmq
mp.messaging.outgoing.quotes.exchange.name=quotes

Note that in this case we have one incoming and one outgoing connector configuration, each one distinctly named. The configuration properties are structured as follows:

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

channel-name 片段必须与 @Incoming 和 @Outgoing 注解中设定的值相匹配:

  • quote-requests → RabbitMQ queue from which we read the quote requests

  • quotes → RabbitMQ exchange in which we write the quotes

接收报价

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

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 使用 @Channel 修饰符注入 quotes 通道
2 表示内容是使用 Server Sent Events 发送的
3 返回流 (Reactive Stream) 。

Again we need to configure the incoming quotes channel inside producer project. Add the following inside application.properties file:

# Configure the outgoing `quote-requests` queue
mp.messaging.outgoing.quote-requests.connector=smallrye-rabbitmq

# Configure the incoming `quotes` queue
mp.messaging.incoming.quotes.connector=smallrye-rabbitmq

HTML页面

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

在 producer 项目中创建 src/main/resources/META-INF/resources/quotes.html 文件，内容如下：

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

这里没有什么特别之处。对于每一个收到的报价，它都会更新页面。

让它运行起来

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

mvn -f rabbitmq-quickstart-producer quarkus:dev

并且，在另一台终端中：

mvn -f rabbitmq-quickstart-processor quarkus:dev

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

在你的浏览器中打开 http://localhost:8080/quotes.html ，点击按钮来请求一些报价。

在JVM或本地模式下运行

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

version: '2'

services:

  rabbit:
    image: rabbitmq:3.12-management
    ports:
      - "5672:5672"
    networks:
      - rabbitmq-quickstart-network

  producer:
    image: quarkus-quickstarts/rabbitmq-quickstart-producer:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: rabbitmq-quickstart-producer
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      RABBITMQ_HOST: rabbit
      RABBITMQ_PORT: 5672
    ports:
      - "8080:8080"
    networks:
      - rabbitmq-quickstart-network

  processor:
    image: quarkus-quickstarts/rabbitmq-quickstart-processor:1.0-${QUARKUS_MODE:-jvm}
    build:
      context: rabbitmq-quickstart-processor
      dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm}
    environment:
      RABBITMQ_HOST: rabbit
      RABBITMQ_PORT: 5672
    networks:
      - rabbitmq-quickstart-network

networks:
  rabbitmq-quickstart-network:
    name: rabbitmq-quickstart

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

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

mvn -f rabbitmq-quickstart-producer clean package
mvn -f rabbitmq-quickstart-processor clean package

一旦打包完成，请运行 docker compose up --build 。UI在 http://localhost:8080/quotes.html

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

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

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

export QUARKUS_MODE=native
docker compose up --build

与之前一样，UI也是在 http://localhost:8080/quotes.html

进一步探索

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

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

Related content

On the same extensions

On the same topics

Quarkus is open. All dependencies of this project are available under the Apache Software License 2.0 or compatible license.

This website was built with Jekyll, is hosted on GitHub Pages and is completely open source. If you want to make it better, fork the website and show us what you’ve got.

Navigation
Follow Us
Get Help
Languages
Quarkus is made of community projects
CC by 3.0 | Privacy Policy Sponsored by