第 5 章：微服务间的通信

本章内容概览：

利用应用级的实时重载功能，加速迭代过程
通过 HTTP 请求实现微服务间的直接消息传递
通过 RabbitMQ 实现微服务间的间接消息传递
对比直接消息和间接消息的使用场景

微服务应用由众多微服务构成，每个微服务负责特定的功能领域。鉴于单个微服务的任务较为简单，微服务之间需要相互协作，以完成整个产品功能集所需的复杂行为。这要求微服务之间必须能够进行有效通信；如果它们无法交流，将无法协调各自的活动，从而无法实现有效的成果。

在本章，我们将探讨微服务如何通过通信协作，以满足应用程序的高级需求。我们还将重新审视 Docker 和 Docker Compose 的使用，以设置整个应用的实时重载，从而避免每次代码更新时都需要重新构建和启动应用。

前面章节展示了 HTTP 请求是微服务间通信的一种方式。本章将深入探讨如何通过 HTTP 请求进行直接消息传递，并研究利用 RabbitMQ 进行间接消息传递。通过本章的学习，你将能够根据具体情境选择合适的消息传递方法。

5.1 新工具与熟悉的工具

本章将介绍 RabbitMQ 这一消息队列软件，它帮助我们实现微服务的解耦。我们将使用名为amqplib的 npm 包，以实现微服务与 RabbitMQ 之间的连接，从而进行消息的发送与接收。同时，我们也将回顾一些熟悉的工具，并探讨如何通过 HTTP 请求发送消息以及如何升级开发环境以支持应用级的实时重载。本章介绍的工具列表如下表所示。

表 5.1 新工具与熟悉工具

工具	版本	描述
Docker Compose	24.0.5	通过 Docker Compose，我们可以同时配置、构建、运行和管理多个容器。
HTTP	1.1	使用 HTTP 进行微服务间的直接（或同步）消息传递。
RabbitMQ	3.12.4	使用 RabbitMQ 作为发送间接（或异步）消息的消息队列软件。
amqplib	0.10.3	该 npm 包允许我们配置 RabbitMQ，并通过 JavaScript 发送和接收消息。

5.2 获取代码

要按照本章内容进行操作，请下载代码或克隆仓库。从以下链接下载代码的 zip 文件： http://mng.bz/vPM1 。你也可以使用 Git 克隆代码：

git clone https://github.com/bootstrapping-microservices-2nd-edition/chapter-5

如需安装和使用 Git 的帮助，请参见第 2 章。如果在代码中遇到问题，请在 GitHub 上针对仓库提交问题。

5.3 让我们的微服务开始对话

到目前为止，我们的应用包含两个微服务：视频流和视频存储。在上一章，我们添加了数据存储功能，现在视频流微服务配备了数据库，而视频存储微服务则使用外部云存储保存视频文件。图 5.1 展示了当前应用的架构。

图 5.1 在上一章的基础上，我们的应用现在包括两个微服务和一个数据库，均在开发计算机上通过 Docker Compose 运行。我们还连接了 Azure Storage 作为外部云存储来存储视频。

微服务应用的构建依赖于各个服务的协作以提供应用功能。如果微服务间无法通信，整个应用将无法有效执行任何操作。因此，微服务间的通信是构建微服务架构的关键组成部分，必须确保它们能够有效地协作。

实际上，如果不是我们在第 4 章已经实现了通过 HTTP 请求进行通信，我们无法达到现在的进展。尽管之前并未详细讨论，但这一点至关重要。如果没有它，我们的应用在实现流媒体和存储分离的第一个挑战上就会失败。

注意：我们的微服务必须协同工作以实现应用功能，因此它们之间的通信非常关键。

在本章，我们将为应用添加第三个微服务：历史记录微服务。引入这个新微服务的目的是演示微服务间的通信方式。如图 5.2 所示，你可以看到视频流微服务是如何向历史记录微服务发送消息流的。

图 5.2 本章中，我们通过引入一个新的微服务来扩展应用，并探索微服务间的通信方法。

5.4 介绍历史微服务

本章中，我们以历史微服务为例，展示微服务之间如何进行消息的发送与接收。事实上，这个新的微服务在 FlixTube 中扮演了一个重要的角色，顾名思义，它负责记录用户的观看历史。

我们的应用程序可以多方面利用这些观看记录。首先，用户可以查看自己的历史记录，回顾之前观看过的视频或找到之前暂停的视频继续观看。此外，观看历史还可以帮助我们为用户提供视频推荐。

为了简化本章的示例，我们将从系统中移除视频存储微服务，从而简化视频流微服务的结构。实际上，作为本章的开始，我们会回退到视频流微服务的早期版本，这个版本将视频直接内嵌在其 Docker 镜像中。我们将继续使用第 3 章后的视频流微服务设置。这种简化只是为了在我们理解通信技术之前的临时措施。在本章结束后，我们会重新引入视频存储微服务，并恢复视频流微服务到其原始状态。

在微服务之间传递的主要消息是“已观看”消息，这是视频流微服务用来通知历史微服务用户已观看某视频的方法。图 5.3 向你展示了历史微服务的工作内容：接收来自视频流微服务的消息并将其记录在自己的数据库中。

图 5.3 作为探索沟通方法的一种方式，我们将让视频流微服务向历史微服务发送已观看消息以记录我们用户的观看历史。

我们还未讨论可采用的消息传递方式——这将在后续部分介绍。目前，请知道我们有多种技术可以发送“已观看”消息。通过本章的学习，我们将探索各种选项，并在以后决定哪种方法最适合特定的应用场景。但首先，让我们升级我们的开发环境，以实现更快的开发周期。

5.5 快速迭代的实时重载

在第 2 章的 2.4 节，我们讨论了我们的开发哲学，即小步快跑、频繁迭代是紧密反馈循环和快速开发速度的关键。在第 2 章中，我们直接在 Node.js 环境下运行第一个微服务，使用 npm 包 nodemon 实现了微服务的实时重载。这意味着在我们修改代码时，微服务会自动重新加载。在整个应用程序级别实现高效的实时重载机制比在单个微服务级别更为重要，因为构建和启动由多个微服务组成的整个应用程序比启动单个微服务要慢得多。

在第 3 章，我们开始使用 Docker，并将我们的微服务代码“烧录”到 Docker 镜像中。Docker 对我们来说是一种非常有用的工具，用于打包、发布和部署我们的微服务。这就是为什么我们选择使用它，尽管我们还未涉及到部署的具体细节；这一点将在第 6 章中展开。

第 4 章中，我们在开发环境中引入了 Docker Compose，这是一种便捷的方法来构建和管理我们不断增长的应用程序的本地版本。这一切都很好，但不幸的是，当我们从直接使用 Node.js 转为在 Docker 容器中运行微服务时，我们失去了代码自动重载的功能。

因为我们将代码烧录到 Docker 镜像中，一旦完成就无法更改！这对生产环境来说是理想的，因为出于安全原因，许多公司希望能够验证软件供应链，并确保没有人在镜像中注入任何（潜在的恶意）代码。现在的问题是，在开发过程中，我们不想不断重建镜像和重启应用程序以包括更新的代码。这个过程相当缓慢。频繁的重建和重启会逐渐消耗时间，尤其是当应用程序规模扩大时。

注意：在运行中的应用程序中不能快速更新代码，这对我们的开发过程是一个严重的障碍，可能会大幅降低我们的生产效率。我们现在将解决这个问题，并找到一种方法恢复我们的实时重载功能。

在本节中，我们将升级我们的 Docker Compose 文件，以支持在开发计算机和容器之间共享代码。图 5.4 向你展示了新历史微服务的源代码目录如何从我们的开发计算机共享到容器中。

我们将再次利用 nodemon 来实现这一功能。它将在代码更改时自动重启每个微服务。虽然这种配置可能看起来复杂，但它将大大提高我们的开发效率，因此非常重要确保正确设置！

5.5.1 创建历史微服务的框架

首先，我们将为新的历史微服务配置实时重载，然后将这种配置扩展到所有微服务。这样，应用程序中的每个微服务都将支持实时重载。

在开始之前，请查阅清单 5.1（chapter-5/example-1/history/src/index.js）以了解新建的历史微服务框架。目前，它还没有具备实际功能，只是一个待填充功能的框架。设置好实时重载后，我们可以使用 Docker Compose 启动应用程序，并进行实时更新和逐步改进，以完善这个新微服务。

清单 5.1 历史微服务的框架

const express = require("express"); // 引入 Express 库

--snip--

const PORT = process.env.PORT;

async function main() { // 这是一个框架微服务。稍后，我们将在此添加 HTTP 路由和消息处理程序。
  const app = express();

  // ... 在此添加路由处理程序 ...

  app.listen(PORT, () => {
    console.log("微服务启动成功。");
  });
}

main()
  .catch(err => {
    console.error("微服务启动异常。");
    console.error(err && err.stack || err);
  });

5.5.2 增强微服务以支持实时重载

为了支持实时重载，我们无需对微服务的基础代码进行其他修改，仅需按第 2 章所学，安装 nodemon。每个微服务都应安装 nodemon：

npm install --save-dev nodemon

我们将使用 nodemon 监控代码变动，并在代码有更改时自动重启微服务。微服务的 package.json 文件中将包含一个名为 start:dev 的 npm 脚本，这一设置从第 2 章开始就已经使用。你可以在清单 5.2（chapter-5/example-1/history/package.json）中看到其配置示例。

清单 5.2 使用 nodemon 配置 package.json 以支持实时重载

{
  "name": "history",
  "version": "1.0.0",
  "description": "",
  "main": "./src/index.js",
  "scripts": {
    "start": "node ./src/index.js",
    "start:dev": "nodemon --legacy-watch ./src/index.js" // 使用 nodemon 实现实时重载，代码变更时，nodemon 会自动重启微服务。
  },
  --snip-- 
  "dependencies": {
    "express": "^5.0.0-beta.1"
  },
  "devDependencies": {
    "nodemon": "^2.0.21"
  }
}

有了 start:dev 脚本，我们可以如下运行微服务：

npm run start:dev

这将调用 nodemon 启动微服务：

nodemon --legacy-watch ./src/index.js

虽然可以随时手动输入完整的 nodemon 命令，但使用 npm run start:dev 更为简洁，并保持所有微服务的一致性（假设我们对每个微服务都应用了这种做法）。如果你已经启动了历史微服务，请使用 Ctrl-C 退出。我们很快将通过 Docker Compose 运行整个应用程序。

你可能会问，为什么在 nodemon 命令中使用了 –legacy-watch 参数。这是因为我通常在 Windows 上的 WSL2 中运行 Docker 和 Docker Compose。--legacy-watch 参数禁用了文件系统监控，改为使用轮询机制来监测代码变化。如果你在 WSL2 或虚拟机（VM）中开发，则需要此参数，因为自动文件监控不会从主机操作系统传递更改。如果你不在 WSL2 或 VM 中开发，可以安全地移除 –legacy-watch 参数，以获得更好的性能。

5.5.3 区分开发和生产的 Dockerfile

在第 2 章中，我们讨论了如何区分开发模式和生产模式下的微服务运行。这种区分是为了针对不同需求进行优化：开发模式优先考虑快速迭代，而生产模式则优先考虑性能和安全性。

因此，对于所有微服务，我们将创建两个 Dockerfile：一个用于开发，另一个用于生产。开发版本命名为 Dockerfile-dev，生产版本命名为 Dockerfile-prod。

这种命名是为了清晰和避免混淆。在软件开发中，清晰的命名非常重要，可以避免误解。分离 Dockerfile 的目的是在开发中启用实时重载功能，这在生产中通常不需要启用。

清单 5.3（chapter-5/example-1/history/Dockerfile-prod）展示了新的历史微服务的生产 Dockerfile。这是一个标准的 Node.js Dockerfile，与第 3 章中创建的类似。

清单 5.3 创建生产 Dockerfile

FROM node:18.5.0

WORKDIR /usr/src/app
COPY package*.json ./
RUN npm ci --omit=dev // 仅安装生产依赖
COPY ./src ./src // 将源代码复制到镜像中，代码被“烧录”到镜像中。

CMD npm start // 以生产模式启动微服务

尽管我们不会在本章中使用生产 Dockerfile，但从第 6 章开始，当部署到生产环境时，将需要这些文件。并行维护开发和生产的 Dockerfile 是一个好策略，以确保开发版本不超前于生产版本。

清单 5.4（chapter-5/example-1/history/Dockerfile-dev）展示了历史微服务的开发 Dockerfile。阅读时，请与清单 5.3 中的生产 Dockerfile 进行比较，了解开发与生产之间的差异。

清单 5.4 创建开发 Dockerfile

FROM node:18.5.0

WORKDIR /usr/src/app
COPY package*.json ./

CMD npm install --prefer-offline && \ // 容器启动时从缓存中安装 npm，使用缓存可以加快安装速度。
    npm run start:dev // 以开发模式启动微服务

两个 Dockerfile 之间的主要区别在于：在清单 5.3 中，我们只安装生产依赖，而在清单 5.4 中，我们安装了所有依赖，包括开发依赖。最重要的区别是清单 5.3 中的 COPY 指令将代码烧录到生产 Docker 镜像中：

COPY ./src ./src

而在开发 Dockerfile 中，没有复制代码的指令。这是因为我们不希望在开发 Docker 镜像中直接包含代码，以便稍后可以轻松更改代码并使用实时重载。

那么代码如何进入容器呢？我们将在下一节中找到答案。同时，开发和生产 Dockerfile 之间还有其他区别需要注意。

请注意在容器内启动微服务的 CMD 指令。在生产 Dockerfile 中，我们使用第 2 章描述的 npm start 约定来启动微服务：

CMD npm start

而开发 Dockerfile 中的 CMD 指令则更为复杂，它执行了更多操作：

CMD npm install --prefer-offline && \ 
npm run start:dev

这个命令被分为两部分，使用反斜杠 () 作为行继续字符。第一部分安装依赖，第二部分启动微服务。

在生产 Dockerfile 中，我们在 Docker 构建过程中调用 npm install，将依赖编译进镜像，这是生产环境的标准操作。而在开发版本中，我们在容器启动时执行 npm install，目标是更好的性能。

通过在容器启动时执行 npm install，我们可以利用主机操作系统上缓存的 npm 包，从而加快后续的 npm 安装和容器启动速度。你将在下一节中了解这是如何实现的。

开发 Dockerfile 中的 CMD 指令的第二部分启动微服务，调用 npm 脚本 start:dev 以实现实时重载。

5.5.4 更新 Docker Compose 文件以支持实时重载

为了实现整个应用程序的实时重载，关键一步是修改 Docker Compose 文件，使宿主操作系统和容器之间能够共享代码和 npm 缓存。本节我们将利用 Docker 卷在开发机和容器间共享文件系统，这意味着你可以在 Visual Studio Code (VS Code) 中编辑代码，并且几乎可以立即看到 Docker Compose 下运行的微服务中的变化。

清单 5.5 展示了 example-1 的 Docker Compose 文件（chapter-5/example-1/docker-compose.yaml）配置，特别是针对新的历史微服务。这与第 4 章创建的 Docker Compose 文件大体相似，但包括了一些修改和新增内容。

清单 5.5 更新 Docker Compose 文件以支持实时重载

version: '3'
services:
  --snip--
  history: # 新历史微服务的容器定义
    image: history
    build:
      context: ./history
      dockerfile: Dockerfile-dev # 使用开发版 Dockerfile
    container_name: history
    volumes: # 在宿主操作系统和容器之间定义共享的卷
      - /tmp/history/npm-cache:/root/.npm:z # 共享 npm 缓存到容器，提升 npm 模块安装速度。
      - ./history/src:/usr/src/app/src:z # 将源代码直接共享到容器，实现代码更改的自动同步。
    ports:
      - "4002:80"
    environment:
      - PORT=80
      - NODE_ENV=development
    restart: "no"

清单 5.5 中的关键更新是使用 Dockerfile-dev，这是开发版的 Dockerfile。尽管在第 4 章我们提到可以省略 dockerfile 字段，默认为 Dockerfile，但这里我们明确设置为 Dockerfile-dev，以便使用开发配置。

新增的 volumes 字段创建了一些 Docker 卷，这些卷将开发机的文件系统与容器的文件系统连接起来。这是我们选择不将代码直接“烧录”到镜像中的原因。

代码共享通过一个 Docker 卷实现，另一个卷则用于创建一个共享 npm 缓存目录。这允许容器中安装的 npm 包在宿主操作系统上进行缓存，从而提速后续的 npm 安装。

你可能会对清单 5.5 中卷配置使用的 z 标志感到好奇，这是一种向 Docker 指示该卷将被共享（可能在多个容器之间）的方式。如需更多信息，可参考 Docker 官方文档： https://docs.docker.com/storage/bind-mounts/ 。

以上步骤仅针对历史微服务！我们需要对所有微服务实施类似的更改。幸运的是，我们可以应用相同的模式到每个微服务，具体如下：

为每个微服务安装 nodemon。
更新 package.json，配置 start:dev 脚本使用 nodemon 启动微服务（参见清单 5.2）。
为开发和生产模式分别创建 Dockerfile。开发版 Dockerfile 不应将代码复制到镜像中（参见清单 5.4）。
仅在开发模式下，在容器启动时执行 npm 安装，而非生产模式（为了优化性能，参见清单 5.4）。
更新 Docker Compose 文件，使用开发版 Dockerfile（参见清单 5.5）。
在 Docker Compose 文件中添加 Docker 卷，共享源代码和 npm 缓存到容器（参见清单 5.5）。

我已经为第 5 章存储库中的所有示例提前完成了这些配置，因此你无需担心。但你至少应该启动 example-1，对历史微服务进行一些代码更改，体验实时重载的效果！现在就让我们这样做吧。

5.5.5 尝试实时重载

不要只是阅读代码清单！现在是时候看看实时重载的实际效果了，这样你才能真正体会到其价值。打开终端，进入第 5 章代码存储库下的 example-1 子目录，然后使用 Docker Compose 启动应用程序：

cd chapter-5/example-1
docker compose up --build

这个示例包括了简化的视频流微服务和新的历史微服务。观察 Docker Compose 的输出，你应该会看到由历史微服务在启动时打印的“Hello world!”消息。为了测试实时重载，我们将更改历史微服务打印的消息：

在 VS Code 中打开 example-1 目录。
找到并打开历史微服务的 index.js 文件。
修改打印“Hello world!”消息的代码行，改为“Hello computer!”。
保存 index.js 文件，然后切换回 Docker Compose 的输出。

如果你动作够快，你会看到历史微服务正在重新加载并显示你更新的消息。如果你反应慢了，你应该会发现这已经发生了。注意视频流微服务没有重新加载，因为我们没有更改其代码。只有历史微服务发生了更新，因此只有它重新加载了。

这就是实时重载的优势。我们可以快速迭代更新代码，并获得即时反馈。我们无需等待构建并启动整个应用程序。相反，我们可以热重载需要更新的每个微服务的代码。

如果我们在代码中引入了错误会怎样？如果一个微服务因错误重新加载，错误将显示在 Docker Compose 的输出中。然后我们可以更正错误并保存代码文件。微服务会自动重新加载，如果我们的更改确实修复了错误，我们应该会看到来自更新后的微服务的干净输出。

此时，我建议你故意尝试破坏历史微服务以看看会发生什么。来吧，打开它的 index.js 文件，并输入一些肯定会引发错误的随机字符。保存文件，然后切换回 Docker Compose 的输出以查看结果。

考虑错误消息的含义以及你所做的操作如何导致了该错误。现在，花点时间尝试破坏代码，制造问题，并在过程中找到一些乐趣。

强制容器重启

如果需要强制重载未更改的微服务——例如微服务已经挂起或崩溃，现在卡住了——我们通常需要更改代码以使容器重启，如添加一些空格然后保存文件。

实际上，我们不需要这样做。我们可以简单地在 VS Code 中保存文件，这足以触发容器重启。我们不需要实际更改代码！

如果你的终端支持 touch 命令，你也可以从命令行为历史微服务触发实时重载，操作如下：

cd chapter-5/example-1
touch history/src/index.js

如果你没有为特定容器设置实时重载（通常只为经常更改的微服务设置），那么你可以使用 Docker Compose restart 命令来使容器重启，例如：

docker compose build history
docker compose restart history

5.5.6 在开发环境中测试生产模式

虽然本章中我们已经分别为开发模式和生产模式准备了不同的 Dockerfile，但目前我们还没有使用生产版本的 Dockerfile。这将在第 6 章改变，届时我们将部署到生产环境。尽管还未部署，这并不意味着我们不能在生产模式中进行本地测试。实际上，我们应该定期在开发环境中测试生产模式。

在开发过程中，我们经常进行小的代码更改并验证应用程序是否依然正常运行。虽然我们使用生产版 Dockerfile 的频率没有开发版本那么高，但这些文件应该与开发版本同步更新。我们也应定期在生产模式下进行测试，尽管这样的测试不必像开发模式测试那么频繁。

比如，我们可能每隔几分钟就在开发模式下进行一次测试，因为代码有所变动。我们也需要测试生产模式，但可能每隔几小时，或在进行了大量代码更新后，才测试一次。关键是，我们需要在本地测试我们的生产 Dockerfile，确保在部署到生产环境前不会发现任何潜在问题。

我们可以通过定期在开发计算机上以生产模式进行测试来轻松解决这一问题。为此，我通常维护两个单独的 Docker Compose 文件：一个用于开发，另一个用于生产。

使用 Docker Compose 时，我们可以通过 -f 参数指定使用哪个 Docker Compose 文件。例如，若要在开发计算机上以生产模式运行应用程序，我们可以创建一个专门的生产版 Docker Compose 文件，并像这样执行：

docker compose -f docker-compose-prod.yml up --build

虽然我们可以使用单个通过环境变量参数化的 Docker Compose 文件，但我倾向于保留用于开发和生产测试的不同版本。这样做的原因是我希望生产 Docker Compose 文件尽可能地模拟真实的生产环境。此外，通常我会在开发版本中使用各种模拟微服务以便于快速简单地测试，但在生产模式测试时，最好使用真实的微服务而非模拟版本。

我们将在第 10 章讨论模拟微服务。而在第 9 章，我们将介绍自动化测试，这也有助于提升你的开发效率。

5.5.7 我们已经实现了什么？

在 5.5 节中，我们配置了微服务以支持实时重载。我们从历史微服务开始，并将相同的配置应用于视频流微服务。从现在开始，我们将为所有微服务采用这种模式。

我们这样做是因为构建和启动我们的整个应用程序通常需要很长时间。我们不想为每一次小的代码更改都重新构建和启动应用程序。相反，我们希望能够迅速地进行代码更改，快速实验和迭代，让应用程序自动更新。现在，我们可以编辑代码，微服务会自动重启。这就是所谓的实时重载——在我们编码的同时自动进行。

这大大提高了工作流的效率和效果。我们现在可以持续发展我们的微服务应用程序，同时接收持续的反馈。浏览 example-1 中的代码，确保你理解实时重载配置是如何整合到整个应用程序中的。

5.6 微服务之间的通信方式

在将开发环境升级为支持整个应用范围的实时重载之后，让我们回到本章的核心主题：探索微服务之间的通信机制。在深入研究通信技术之前，我们将首先概述两种基本的微服务通信方式：直接消息传递和间接消息传递，通常也被称为同步和异步通信。

我更倾向于使用“直接”和“间接”这些术语，而不是“同步”和“异步”，因为在常规计算机编程中后两者有着不同的含义。此外，异步编程尤其是一个难点，对许多有抱负的程序员来说学起来很困难。不用担心；我们会避免使用“异步”这个词。

5.6.1 直接消息传递

直接消息传递指的是一个微服务直接向另一个微服务发送消息，并立即接收到响应。我们使用消息主要有两个目的：通知某个微服务系统事件或触发微服务中的特定操作。任何消息都可以视为通知或命令（或两者的结合）。当我们发送一个命令给另一个微服务，以触发某个操作并立即获得操作成功或失败的反馈时，直接消息传递特别有用。

直接消息传递还可以用来管理多个微服务之间的行为序列。可以将其视为向一系列微服务发送命令（比如，“执行这个操作”，然后“告诉我结果”）。

接收方微服务无法忽略或回避传入的消息。如果尝试这样做，发送方将从响应中立即得知。图 5.5 展示了视频流微服务如何将“已观看”消息直接发送给历史微服务，后者提供了直接且即时的响应。

直接消息传递是某些用例所必需的。它的主要缺点是需要通信两端的微服务之间的紧密耦合。通常，我们更希望避免微服务之间的紧密耦合，因此，通常会优先考虑间接消息传递。

5.6.2 间接消息传递

间接消息传递引入了一个中介层，介于通信的两个端点之间。这意味着通信的双方不需要直接了解对方。这种方式显著降低了微服务之间的耦合度，具体表现在：

消息通过中介发送，使发送者和接收者都不清楚涉及的具体是哪个其他微服务。对于发送者来说，它甚至可能不知道是否有其他任何微服务接收了这个消息。
由于接收者不知道是哪个微服务发送的消息，因此无法直接回复。这意味着这种通信方式不适用于需要直接反馈以确认成功或失败的场景。

当发送的微服务不关心是否采取了后续行动时，我们应该使用间接消息。它也可以用于广播通知，例如，宣布一个重要事件，可能有其他微服务需要知道。

注意我们使用间接消息传递来宣布不需要直接回应的重要事件。这种消息方式允许更灵活的通信结构，并减少微服务之间的耦合。

图 5.6 展示了视频流微服务（左侧）如何通过一个消息队列（中介）向历史微服务（右侧）发送间接消息。请注意，视频流微服务与历史微服务之间没有直接连接，这就是所谓的松散耦合。

图 5.6 间接消息不是直接发送到一个微服务；而是被放置在队列中，稍后可以处理。

间接消息传递可以帮助我们构建灵活的消息架构，解决许多复杂的通信问题。不幸的是，这种灵活性也带来了更高的复杂性。随着应用程序的扩展，你会发现绘制精确的通信路径更加困难，因为它们不再是直接的，因此不那么明显。有了直接和间接消息传递的概述，我们现在可以实际尝试这些通信方法。

5.7 使用 HTTP 进行直接消息传递

在上一章中，我们使用 HTTP 进行数据检索，检索我们的流媒体视频。本章中，我们使用 HTTP 的目的不同：直接从一个微服务发送消息到另一个微服务。

注意使用 HTTP 请求发送的消息有直接的回应。我们可以立即知道消息的处理是否成功或失败。

具体来说，在本节中，我们将使用 HTTP POST 请求直接从视频流微服务发送消息到历史微服务。图 5.7 展示了这一过程。

5.7.1 为什么选择 HTTP？

HTTP 是万维网的语言和基础，也是创建网络服务的事实标准。HTTP 是可靠的，并且广为人知。

由于 HTTP 已经普遍用于创建 REST API，我们不需过多考虑为何选择使用 HTTP。它是为此类用途设计的，且几乎所有编程语言都提供了支持。此外，我们也可以轻松访问大量与 HTTP 相关的资源，这些资源本身就是通过支持万维网的 HTTP 提供的。

5.7.2 直接针对特定微服务发送消息

在向一个微服务发送消息之前，我们需要一种方法来定位它。伴随 HTTP 的是另一种互联网协议，称为域名系统（DNS），它为我们提供了一种简单且自动的方式，通过使用它们的名称直接将消息发送到微服务。

关于如何将消息直接发送到另一个微服务的问题，最简单的答案是使用无处不在的 DNS，它将主机名转换为 IP 地址。这在 Docker Compose 中自动发生（容器名称即是主机名），而在生产的 Kubernetes 集群中，使其工作也不需太多额外努力。

图 5.8 展示了我们如何向特定的主机名发送 HTTP POST 消息。当发送 HTTP 请求时，DNS 查找是自动进行的，它将我们的主机名转换为微服务的 IP 地址。

图 5.8 HTTP 请求通过 DNS 查找路由，将目标微服务的主机名转换为 IP 地址。

IP 地址代表我们微服务在互联网上的唯一位置。注意，这并不一定涉及公共互联网。在这种情况下，IP 地址实际上代表的是私有网络中的服务器，无论是在我们的开发计算机上的 Docker 运行时还是在生产 Kubernetes 集群中操作。我们需要 IP 地址来通过 HTTP 请求直接将消息定向到接收方，而 DNS 几乎像魔法一样在幕后自动工作。

在使用 Docker 和 Docker Compose 开发时，DNS 自动工作，我们可以依赖它。当我们部署到生产 Kubernetes 集群时，我们还需要确保我们的微服务可以通过 DNS 访问，但我们将在第 6 章中解决这个问题。

5.7.3 使用 HTTP POST 发送消息

发送消息的方程有两个方面：一个微服务发送消息，另一个接收消息。在本节中，我们将探讨如何使用 HTTP POST 请求发送消息。

在第 4 章的 4.4.2 节中，我们看到了从一个微服务到另一个微服务的 HTTP GET 请求。我们使用了内置的 Node.js HTTP 库来执行此操作。我们将再次使用这个库从一个微服务发送请求到另一个微服务。

清单 5.6 是从 example-2 视频流微服务的更新后的 index.js 文件中提取的（chapter-5/example-2/video-streaming/src/index.js），展示了如何发送 HTTP POST 消息。它实现了一个新函数 sendViewedMessage，每当用户开始观看视频时，就向历史微服务发送“已观看”的消息。

清单 5.6 使用 HTTP POST 发送直接消息

function sendViewedMessage(videoPath) {                           // 向历史微服务发送“已观看”的消息的辅助函数
  const postOptions = {                                           // 配置 HTTP 请求选项
    method: "POST",                                               // 设置 HTTP 方法为 POST
    headers: {
      "Content-Type": "application/json",                         // 设置内容类型为 JSON
    },
  };

  const requestBody = {                                           // 定义 HTTP 请求的正文，包含消息数据
    videoPath: videoPath
  };

  const req = http.request(                                       // 发送 HTTP 请求
    "http://history/viewed",                                      // 指定请求 URL，标识历史微服务及其“已观看”路由
    postOptions
  );

  req.on("close", () => {                                         // 请求完成时的回调函数
    --snip--
  });

  req.on("error", (err) => {                                      // 处理可能出现的错误
    --snip--
  });

  req.write(JSON.stringify(requestBody));                         // 将消息数据写入请求
  req.end();                                                      // 结束请求
}

我们通过调用 http.request 函数创建 HTTP POST 请求。我们通过 URL http://history/viewed 将请求定向到历史微服务，该 URL 包括主机名（history）和路径（viewed）。这种组合标识了目标微服务及我们向其发送的消息。

单独的回调函数处理请求成功与失败。在此处，我们可以捕获错误并采取后续措施。否则，如果请求成功，我们可能希望执行后续操作。

5.7.4 使用 HTTP POST 接收消息

在方程的另一侧，我们通过在接收微服务中定义一个 Express 路由处理程序来接收 HTTP POST 消息。清单 5.7 显示了历史微服务的 index.js 文件的摘录（从 chapter-5/example-2/history/src/index.js 中提取），演示了这一点。新的 HTTP POST 路由处理程序用于接收并处理通过 viewed 路由发送的消息。在这个例子中，我们简单地将接收到的消息存储在数据库中以保存观看历史。

清单 5.7 使用 HTTP POST 接收直接消息

const historyCollection = db.collection("history");

app.post("/viewed", async (req, res) => {                          // 处理接收到的“已观看”消息的 HTTP POST 请求
  const videoPath = req.body.videoPath;                            // 从请求的 JSON 正文中提取视频路径
  await historyCollection.insertOne({                              // 在数据库中插入记录以保存观看信息
    videoPath: videoPath
  });
  res.sendStatus(200);                                             // 发送 HTTP 状态 200，表示成功处理请求
});

注意，在 HTTP POST 处理程序中我们如何通过 req.body 访问请求正文，将其视为消息负载。由于我们使用了 Express 的 JSON 解析中间件，正文变量是从 JSON 格式自动解析的。如果你对如何配置 JSON 解析中间件感兴趣，可以查看代码文件 chapter-5/example-2/history/src/index.js。

5.7.5 测试更新后的应用程序

现在，我们可以测试更新后的代码并观察这种消息传递方式的效果。打开终端，切换到 example-2 目录，并以通常方式启动应用程序：

cd chapter-5/example-2
docker compose up --build

如果遇到容器已存在的错误，可能是因为你没有关闭之前的实例。每次从一个示例切换到另一个时，请使用以下命令关闭应用：

docker compose down

等待微服务启动后，在浏览器中访问 http://localhost:4001/video 。此时，测试视频应该开始播放。

然后，返回终端查看 Docker Compose 的输出。你应该会看到视频流微服务发送了一个“已观看”的消息，并随后显示历史微服务接收到该消息的文本。

现在，我们可以直接检查数据库，确保“已观看”的消息已被妥善存储。你需要安装一个数据库查看器，比如 MongoDB Shell 或 Studio 3T（参见第 4 章，4.5.2 节）。

连接数据库查看器到数据库（连接到 Docker Compose 文件中配置的 localhost:4000 端口），检查历史数据库的 videos 集合，确认每次刷新浏览器时都有新记录被创建。检查数据库是测试代码效果的一种直接且实用的方法。

5.7.6 通过直接消息协调行为

直接消息传递的潜在好处之一是可以有一个控制器微服务，在多个其他微服务之间协调复杂的行为序列。直接消息具有即时响应，允许单个微服务直接控制其他微服务的活动。

这种类型的消息被称为同步通信的原因是，我们能够同步地协调消息，如图 5.9 所示。在该图中，微服务 A 正在协调其他微服务的活动。

注意直接消息传递可以用来以明确的方式或明确的顺序协调行为。

使用直接消息，追踪代码和理解消息序列变得容易。你会发现，追踪间接消息的序列并不容易。

图 5.9 直接消息传递允许一个控制器微服务（这里是微服务 A）在多个其他微服务之间协调复杂行为。

5.7.7 我们实现了什么？

在 5.7 节中，我们探讨了使用 HTTP POST 请求从一个微服务直接发送“已观看”消息到另一个微服务的方法。这种方式被称为直接消息传递，因为我们可以通过名称直接将消息定向到特定的微服务。我们还能立即得知消息处理的成功或失败。

这类消息更应被视为命令或行动的呼吁，而不仅仅是通知。由于直接消息的同步性质，我们可以顺序地协调多个消息。这在我们需要一个控制器微服务来协调其他微服务中的复杂行为时非常有用。

尽管直接消息有其用途，有时也是必需的，但它们也存在一些主要缺点。首先，我们一次只能针对单个其他微服务发送消息。因此，当我们希望单个消息被多个接收者接收时，直接消息就不那么适用。

此外，直接消息在微服务之间是一个高耦合点。有时候这种高耦合是必需的，但我们尽可能避免。虽然从控制器微服务集中协调多个微服务的能力可能看起来是一个优势，它确实可以使我们更容易地了解应用程序中正在发生的事情。但最大的问题是，这为可能是大型和复杂的操作创建了单点故障。如果控制微服务在协调过程中崩溃怎么办？我们的应用程序现在可能处于不一致的状态，数据可能已经丢失。直接消息传递带来的问题可以通过间接消息传递来解决，这就是为什么我们现在转向 RabbitMQ。

5.8 使用 RabbitMQ 进行间接消息传递

在掌握了使用 HTTP POST 请求发送直接消息之后，现在是时候探索间接消息传递了，它能帮助我们解耦微服务。虽然这可能使我们的应用程序架构更难理解，但它为安全性、可扩展性、可靠性和性能带来了许多积极影响。

注意 RabbitMQ 允许我们将消息发送者与消息接收者解耦。发送者不知道是否有其他微服务会处理消息。

图 5.10 展示了在添加了 RabbitMQ 服务器后我们应用程序的结构。视频流微服务不再直接与历史微服务耦合。相反，它正在将其“已观看”的消息发布到消息队列中。然后，历史微服务在自己的时间里从队列中拉取消息。

5.8.1 为什么选择 RabbitMQ？

RabbitMQ 是一种广为人知且成熟的消息队列软件。许多公司常用 RabbitMQ，它是用于间接消息传递的首选解决方案。RabbitMQ 开发于十多年前，稳定且成熟。除其他协议外，它还实现了高级消息队列协议（AMQP），这是消息代理通信的一个开放标准。

注意 RabbitMQ 因间接微服务间通信而闻名，并且它允许构建复杂和灵活的消息架构。

RabbitMQ 为所有流行的编程语言提供库，无论你的技术栈是什么，使用它都不会有问题。我们正在使用 Node.js，因此我们将使用 npm 仓库上可用的 amqplib 库。RabbitMQ 是开源的，入门相对容易。你可以在这里找到服务器的代码： https://github.com/rabbitmq/rabbitmq-server 。

5.8.2 间接定位消息到微服务

在间接消息传递中，我们不是直接针对任何特定的微服务，而是需要将消息定向到某处：RabbitMQ 服务器。在 RabbitMQ 服务器中，我们将消息定向到一个命名的队列或消息交换。队列和交换的组合为我们的消息架构提供了很多灵活性。

注意消息发送者使用 DNS 解析 RabbitMQ 服务器的 IP 地址。然后，消息发送者与 RabbitMQ 通信，将消息发布到特定的命名队列或交换上。接收者也使用 DNS 定位并与 RabbitMQ 服务器通信，从队列中检索消息。发送者和接收者在任何时候都不直接通信。

要将消息发布到队列或交换，我们必须首先在我们的应用程序中配置一个 RabbitMQ 服务器。然后，我们可以使用 AMQP 代码库（称为 amqplib）来发送和接收消息。

DNS 在底层将 RabbitMQ 主机名解析为 IP 地址。现在，我们不再像使用 HTTP POST 请求那样将消息直接定向到特定的微服务，而是将这些消息定向到我们的 RabbitMQ 服务器上的特定队列或交换。

我们将在两个部分中进行间接消息的传递，因此我将使用两张图来说明。我们首先考虑使用队列，然后我们将研究使用交换。图 5.11 展示了视频流微服务如何推送其消息到已观看队列。然后，在图 5.12 中，我们将看到历史微服务如何拉取队列中的消息。

我使用了推送和拉取这两个动词，因为这有助于可视化此事务。就像我们之前使用 HTTP POST 一样，我们可以想象视频流微服务正在将其消息推送到历史微服务上，后者无法选择接受与否。消息被强制推送到历史微服务，不考虑微服务是否有处理它的能力。

通过间接消息传递，历史微服务获得了更多控制权。现在它可以在准备好时从队列中拉取消息。当历史微服务负载过重，无法接收新消息时，它可以自由忽略这些消息，让它们在队列中积累，直到微服务能够处理它们。

5.8.3 创建一个 RabbitMQ 服务器

现在我们将在应用程序中添加一个 RabbitMQ 服务器。RabbitMQ 是用 Erlang 语言编写的，尽管可能曾经有一段时间设置它比较困难，但现在这已不再是问题了！得益于你已经学到的 Docker 和 Docker Compose 技能，现在部署变得非常简单。

清单 5.8 是 example-3 Docker Compose 文件的摘录（chapter-5/example-3/docker-compose.yaml），展示了如何在我们的应用程序中添加一个 RabbitMQ 服务器。这是另一个示例，展示了如何从 Docker Hub 上获取镜像并实例化容器，就像我们之前为 MongoDB 数据库所做的那样。

清单 5.8 将 RabbitMQ 服务器添加到 Docker Compose 文件中

version: '3'
services:
  --snip--
  rabbit:                                      # 定义托管我们 RabbitMQ 服务器的容器
    image: rabbitmq:3.12.4-management          # 使用带有管理控制台的 RabbitMQ 镜像版本
    container_name: rabbit                     # 设置容器的名称，这是我们将用来连接到 RabbitMQ 服务器的名称
    ports:                                     # 配置端口映射从宿主机到容器
      - "5672:5672"
      - "15672:15672"
    restart: always                            # 如果 RabbitMQ 服务器遇到问题，自动重启
  --snip--

5.8.4 调查 RabbitMQ 仪表板

你可能已经注意到了清单 5.8 中的端口配置。端口 5672 是我们即将使用 amqplib 通过 RabbitMQ 发送和接收消息的端口。另一个端口 15672，我们将用它来访问 RabbitMQ 的管理仪表板。

注意 RabbitMQ 的仪表板是一个理解 RabbitMQ 运作方式以及更好地理解应用程序中正在传递的消息的有用工具。

我们从名为 rabbitmq:3.12.4-management 的镜像启动了 RabbitMQ 服务器，因为这个版本带有内置的管理仪表板。仪表板如图 5.13 所示，提供了一种图形化方式来探索应用程序中的消息流。让我们现在尝试使用它。

打开一个终端，并切换到 example-3 目录。以常规方式启动应用程序：

cd chapter-5/example-3
docker compose up --build

等待 RabbitMQ 服务器启动，然后在浏览器中访问 http://localhost:15672/ 。使用默认用户名 guest 和默认密码 guest 登录。

你现在应该可以看到 RabbitMQ 仪表板。与图 5.13 不同，如果你还没有创建队列，你可能还不会看到任何队列或交换。这张截图是在创建了 viewed 队列之后拍摄的。我们很快就会触发队列的创建，然后你可以返回仪表板查看它的实际情况。

RabbitMQ 仪表板是一个有价值的调试工具。能够可视化当前发生的事件总是比假设我们知道发生了什么更好。它是那些伟大的可视化工具之一，可以明确显示我们的应用程序实际在做什么。

你可能已经注意到我们可以选择不使用带有仪表板的 RabbitMQ 镜像。我们可以改用 rabbitmq:3.12.4 这个不包含仪表板的镜像版本。如果你正在构建一个需要精简、高效或具有特定安全要求的生产应用程序，这可能是更好的选择。然而，通常情况下，我建议即使在生产环境中也保留仪表板（当然是放在私有网络后面），因为拥有这些工具来帮助我们了解生产环境中发生的事情是非常宝贵的。

5.8.5 将我们的微服务连接到消息队列

现在我们已经配置了一个 RabbitMQ 服务器，接下来就是将我们的微服务连接到它。如果你是从头开始编程，你需要在每个需要连接到 RabbitMQ 的微服务中安装 amqplib npm 包：

npm install --save amqplib

如果你正在直接运行 example-3 的 Node.js 代码，首先需要安装所有依赖：

npm install

列表 5.9 是历史微服务的 index.js 文件的摘录（chapter-5/example-3/history/src/index.js）。它展示了我们如何连接到 RabbitMQ 服务器。

列表 5.9 连接到 RabbitMQ 服务器

--snip--

const amqp = require("amqplib");                      // 导入 amqplib 库，用于与 RabbitMQ 服务器通信

const RABBIT = process.env.RABBIT;                    // 从环境变量获取 RabbitMQ 的连接 URI

async function main() {
  --snip--

  const messagingConnection = 
    await amqp.connect(RABBIT);                       // 连接到 RabbitMQ 服务器
  const messageChannel = 
    await messagingConnection.createChannel();        // 创建一个消息通道

  --snip--

  app.listen(PORT);                                   // 启动 HTTP 服务器
}

列表 5.9 和接下来的列表 5.10 中的关键之一是 RABBIT 环境变量，它配置了连接到 RabbitMQ 服务器的方式。列表 5.10 是 example-3 的 Docker Compose 文件摘录（chapter-5/example-3/docker-compose.yaml）。它设置了 RABBIT 环境变量，包括用户名（guest）、密码（也是 guest）、服务器的主机名（rabbit）和连接的端口号（5672）。

注意在生产环境中，你应该选择一个更安全的用户名和密码，而不是默认值，但在这本书的上下文中，这是足够的，因为我们是在一个私有且可信的网络上运行 RabbitMQ。

列表 5.10 配置历史微服务

version: '3'
services:
  --snip--
  history:
    image: history
    build:
      context: ./history
      dockerfile: Dockerfile-dev
    container_name: history
    volumes:
      - /tmp/history/npm-cache:/root/.npm:z
      - ./history/src:/usr/src/app/src:z
    ports:
      - "4002:80"
    environment:
      - PORT=80
      - RABBIT=amqp://guest:guest@rabbit:5672       # 配置连接到 RabbitMQ 的 URI
      - DBHOST=mongodb://db:27017
      - DBNAME=history
      - NODE_ENV=development
    depends_on:
      - db
      - rabbit                                    # 确保历史微服务依赖于我们在列表 5.8 中定义的 rabbit 容器
    restart: "no"
  --snip--

在你启动这个版本的应用程序之前，有一个问题可能还没有考虑到。RabbitMQ 服务器相对较重，并且启动需要一些时间。与此同时，我们的微服务非常轻量且启动迅速。当微服务尝试连接到尚未准备好的 RabbitMQ 时会发生什么呢？它会因为连接失败而崩溃。现在我们面临的问题是，应用程序中有一些需要按特定顺序解决的启动依赖性。

为了确保微服务能在 RabbitMQ 服务器准备好后再尝试连接，更理想的做法是，如果 RabbitMQ 服务器出现故障（比如，正在进行升级），我们希望微服务能够处理断开的连接并自动重新连接——虽然这更复杂。目前，我们可以采用一个简单的方法来解决这个问题。在第 11 章中，你将学习一种更复杂的方法来处理这个问题。

一个简单的解决方法是在我们的 Dockerfile 中使用一个额外的命令，延迟微服务的启动，直到 RabbitMQ 服务器准备好为止。我们将使用 npm 安装的便捷命令 wait-port 来实现这一点：

npm install --save wait-port

列表 5.11（chapter-5/example-3/history/Dockerfile-dev）展示了更新后的历史微服务的 Dockerfile，其中添加了 wait-port 命令。我们使用这个命令延迟微服务的启动，直到 RabbitMQ 启动后。

列表 5.11 延迟历史微服务等待 RabbitMQ

FROM node:18.17.1

WORKDIR /usr/src/app
COPY package*.json ./

CMD npm install --prefer-offline && \                # 使用 npx 调用本地安装的 wait-port 命令等待，直到主机名 rabbit 在端口 5672 上接受连接
    npx wait-port rabbit:5672 && \
    npm run start:dev                                # 在 wait-port 命令完成后启动历史微服务

同时，我们应该更新生产版本的 Dockerfile。在我们工作时保持两个版本同步是好的做法。

5.8.6 多接收者消息

单一接收者消息是 RabbitMQ 的一个基础用例，也是最容易理解的——这也是为什么我们从它开始。然而，多接收者（或广播式）消息可能更有实际用途。简而言之，一条消息由一个微服务发送，但多个其他微服务可以接收此消息。

这种类型的消息通常用于通知——例如，一条消息表明应用程序中发生了重要事件，如视频已被观看。这是多个微服务可能想要知晓的信息。

请注意，多接收者消息是一对多的：一条消息由一个微服务发送，但可能被多个其他微服务接收。这是在应用内发布通知的一种有效方式。

为了实现这一功能，我们现在需要使用 RabbitMQ 的消息交换功能。图 5.14 展示了视频流微服务如何将“已观看”的消息发布到一个名为“Viewed”的交换。消息随后被路由到多个匿名队列，由多个微服务同时处理。

5.8.7 接收多接收者消息

接收多接收者消息的过程与接收单一接收者消息类似。列表 5.14 是历史微服务的 index.js 文件的摘录（chapter-5/example-4/history/src/index.js），显示了如何从 RabbitMQ 的交换中接收消息。

列表 5.14 从 RabbitMQ 交换消费“已观看”消息

--snip--

const historyCollection = db.collection("history");

await messageChannel.assertExchange("viewed", "fanout");           // 断言存在名为“已观看”的消息交换。选择"fanout"类型将消息传播到所有绑定的队列。

const { queue } = 
  await messageChannel.assertQueue("", { exclusive: true });       // 创建一个匿名队列，并设置为独占，以便只有此连接可以访问它。

await messageChannel.bindQueue(queue, "viewed", "");               // 将队列绑定到交换机

await messageChannel.consume(queue, async (msg) => {               // 从队列中消费消息
  const parsedMsg = JSON.parse(msg.content.toString());            // 解析消息内容
  await historyCollection.insertOne({                              // 在数据库中记录观看信息
    videoPath: parsedMsg.videoPath
  });
  messageChannel.ack(msg);                                         // 确认消息已处理
});

--snip--

5.8.8 发送多接收者消息

发送多接收者消息的过程与发送单一接收者消息类似。列表 5.15 是视频流微服务的 index.js 文件的摘录（chapter-4/example-4/video-streaming/src/index.js），展示了如何发布消息到 RabbitMQ 的交换。

列表 5.15 将“已观看”消息发布到 RabbitMQ 交换

--snip--

async function main() {
  const messagingConnection = await amqp.connect(RABBIT);

  const messageChannel = await messagingConnection.createChannel();

  await messageChannel.assertExchange(
    "viewed", 
    "fanout"                                        // 确保存在名为“已观看”的交换，类型为"fanout"
  );

  function broadcastViewedMessage(messageChannel, videoPath) {    // 定义一个发送“已观看”消息的函数
    const msg = { videoPath: videoPath };                         // 创建消息载体
    const jsonMsg = JSON.stringify(msg);                          // 将消息转换为 JSON 格式
    messageChannel.publish(
      "viewed", 
      "", 
      Buffer.from(jsonMsg)                                        // 将消息发布到交换
    );
  }

  app.get("/video", async (req, res) => {
    --snip--

    broadcastViewedMessage(messageChannel, videoPath);            // 发送观看消息
  });

  --snip--
}

5.8.9 测试多接收者消息

让我们来测试更新后的代码。这次测试中，我引入了一个新的推荐微服务，虽然它目前仅作为一个存根存在，不执行任何操作除了打印出接收到的消息。这足以演示多个微服务如何处理同一消息。打开终端，切换到 example-4 目录，按常规操作执行：

cd chapter-5/example-4
docker compose up --build

启动可能需要一些时间。一旦启动完成，在浏览器中访问 http://localhost:4001/video ，你应该能在控制台看到显示历史微服务和推荐微服务均已接收到“已观看”消息的打印信息。

这种情况的实现是因为我们设置了一个交换机，它绑定到了两个队列——每个接收微服务都有自己的队列。这种方法与使用单一队列不同，在那种情况下，多个微服务将竞争来成为第一个拉取并处理消息的服务，这种方式更类似于负载均衡。虽然这有时候是有用的，但广播式消息通常更加实用。

5.8.8 通过间接消息实现复杂行为

间接消息带来了许多积极的好处，但它们也使得理解和控制应用程序行为变得更加困难。对于间接消息，我们无法获得直接响应，从发送者的视角来看，接收者甚至可能不存在！发送者无法知道是否有任何接收者在等待接收其消息。

注意由于没有对间接消息的中央控制，这促使构建更加灵活、可扩展和可进化的消息架构成为可能。每个微服务负责决定如何响应传入消息，并可能作出响应或生成多个新消息。

与直接消息不同，没有单个微服务负责协调其他微服务的行为。因此，应用程序的行为是从微服务间的间接消息交互自然而然地显现，而不是由一套受控的、明确的指令集定义。这不一定是坏事。拥有单一控制微服务的方法意味着我们有一个潜在的单点故障。如果这个控制微服务在协调过程中崩溃，正在进行的任何活动都可能被中断，导致数据丢失或状态不一致——这是直接消息传递潜在的副作用。

直接消息在某些情况下确实有用，但间接消息通常允许构建更复杂且更具弹性的行为网络。虽然我们可能难以完全理解所有这些如何在其复杂性中结合在一起，但至少我们知道它能够提供高可靠性。这是因为，使用间接消息时，由于微服务之间的连接是通过可靠且容错的消息队列实现的，因此不存在单一故障点（当然，RabbitMQ 也可能失败，但其失败的可能性远小于我们自己的微服务）。

任何特定的微服务都可能失败，但即使它在处理消息时崩溃，我们也知道消息不会丢失，因为当微服务崩溃时消息不会被确认，这些消息最终会被其他微服务（通常是崩溃的那个的副本）处理。这种小巧的技巧有助于构建坚固可靠的微服务应用程序。图 5.15 提供了一个直观视图，展示了间接消息如何在你的应用程序中被编排成动态的消息流。

5.8.9 我们实现了什么？

在本节中，你已经学习了如何通过使用 RabbitMQ 在微服务之间发送间接消息。首先，我们尝试发送单一接收者消息。然后，我们开始发送多接收者消息，以便我们可以在整个应用程序中广播消息。

注意，对于“已观看”消息，使用间接的多接收者消息似乎是正确的选择，因为这样可以减少微服务间的耦合。这是一个显著的胜利。

我们本可以从一开始就计划使用间接的广播式消息，但通过这样的实践，你已经积累了经验，并且现在更能够根据自己的情况决定添加更多消息时需要什么样的消息传递方式。

5.9 微服务通信总结

至此，你已经掌握了两种不同的消息传递方法，使微服务之间能够相互通信。你已经学会了如何通过 HTTP 请求发送直接消息，以及如何使用 RabbitMQ 来发送间接消息。使用 RabbitMQ，你可以进行单一接收者和多接收者（即广播）消息传递。

这为我们的消息架构提供了扩展的灵活性。随着应用程序的扩展，我们可能会添加更多微服务，这些服务可能对“已观看”消息感兴趣或不感兴趣。对于那些感兴趣的微服务，它们可以简单地处理这些消息，而无需我们修改消息的原始发送方。

我们也探讨了在不同情况下选择不同消息传递方式的原因。为了方便参考，这些信息已在表 5.2 中总结。

表 5.2 何时使用每种通信类型

情况	推荐使用
需要将消息直接定向到一个特定微服务	直接消息传递：HTTP
需要确认消息处理是否成功或失败	直接消息传递：HTTP
需要顺序发送多条消息	直接消息传递：HTTP
希望一个微服务来编排其他微服务的活动	直接消息传递：HTTP
需要广播一个消息，通知一个或多个微服务系统中的事件	间接消息传递：RabbitMQ
希望解耦消息的发送者和接收者	间接消息传递：RabbitMQ
发送者和接收者的性能需独立	间接消息传递：RabbitMQ
确保消息处理失败时，能自动重试，直到成功	间接消息传递：RabbitMQ
需要负载平衡处理消息	HTTP 或 RabbitMQ
需要将消息处理分发给多个可以并行行动的接收者	间接消息传递：RabbitMQ

5.10 继续你的学习

本章介绍了如何使用 HTTP 进行直接消息传递，以及如何使用 RabbitMQ 进行间接消息传递。

API Design Patterns 作者 JJ Geewax（Manning，2021）
The Design of Web APIs 作者 Arnaud Lauret（Manning，2019）
RabbitMQ in Depth 作者 Gavin M. Roy（Manning，2017）
RabbitMQ in Action: Distributed Messaging for Everyone 作者 Alvaro Videla 和 Jason J. W. Williams（Manning，2012）

欲了解更多关于 amqplib 包的信息，请参阅文档： https://amqp-node.github.io/amqplib/ 。了解更多关于 wait-port 命令的信息，请查看 GitHub： https://github.com/dwmkerr/wait-port 。

至此，我们已经建立了基础，准备将应用程序推向生产环境。下一章将介绍如何把应用部署到生产环境中。

总结

我们使用 Docker 卷在开发计算机和应用程序容器间共享代码。
使用 nodemon 实现实时重载，更新代码后自动重新加载应用程序的相关微服务。
微服务间有两种通信方式：直接和间接。
直接（即同步）消息传递适用于需要顺序消息流或需要精确编排其他微服务行为的场景。
直接消息传递允许立即知道消息处理是否成功。
间接（即异步）消息传递帮助解耦微服务，促进灵活和可演化应用程序的开发。
使用间接消息，我们可以广播消息，通知整个应用中的重要事件。
HTTP POST 请求用于微服务间直接消息传递。

发布于: 2025/10/25 • 最后更新: 2025/10/26 • 字数: 18711 • 阅读时间: 38 mins