消息驱动式微服务：Spring Cloud Stream & RabbitMQ

1. 概述

在本文中，我们将向您介绍

Spring Cloud Stream

RabbitMQ

Apache Kafka

Spring Cloud Stream

Spring Messaging

Spring Integration

message broker

Spring Cloud Stream

Spring Boot

抽象地讲，让Spring根据配置和依赖关系管理在运行时找出实现自动注入

message broker

本文将使用

RabbitMQ

message broker

broker

2. 微服务中的消息

在微服务体系架构中，我们有许多相互通信以完成请求的小型应用程序—它们的主要优点之一是改进了的可伸缩性。一个请求从多个下游微服务传递到完成是很常见的。例如，假设我们有一个

Service-A

Service-B

Service-C

是的，还会有其他组件，比如

Spring Cloud Eureka

Spring Cloud Zuul

假设由于某种原因

Service-B

I/O操作

DB事务

Service-B

现在，我们可以启动更多的

Service-B

Service-A

Service-B

Service-A

Service-A

另一种方法解决类似情况的是使用事件驱动的微服务体系架构。这基本上意味着

Service-A

HTTP

Service-B

Service-C

message broker

Service-B

Service-C

message broker

与依赖HTTP调用的传统微服务体系架构相比，这有许多优点:

• 提高可伸缩性和可靠性——现在我们知道哪些服务是整个应用程序中的真正瓶颈。
• 鼓励松散耦合——

  Service-A

  不需要了解

  Service-B

  和

  Service-C

  。它只需要连接到

  message broker

  并发布事件。事件如何进一步编排取决于代理设置。通过这种方式，

  Service-A

  可以独立地运行，这是微服务的核心概念之一。
• 与遗留系统交互——通常我们不能将所有东西都移动到一个新的技术堆栈中。我们仍然必须使用遗留系统，虽然速度很慢，但是很可靠。

3. RabbitMQ

高级消息队列协议(AMQP)

RabbitMQ

RabbitMQ

AMQP

3.1 RabbitMQ架构设计

因此发布者将消息发布到

RabbitMQ

Exchange

Exchange

Queues

Exchange

routing

Exchange

Queues

bindings

绑定可以有4种类型:

• Direct: 它根据

  routing key

  (路由键)将

  Exchange

  (交换器)类型直接路由到特定的

  Queues

  (队列)。
• Fanout:它将消息路由到绑定

  Exchange

  (交换器)中的所有

  Queues

  (队列)。
• Topic:它根据完全匹配或部分据

  routing key

  (路由键)匹配将消息路由到(0、1或更多)的

  Queues

  (队列)。
• Headers:它类似于

  Topic

  (主题)交换类型，但是它是基

  routing header

  (路由头）而不是

  routing key

  (路由键)来路由的。



来源： https://www.cloudamqp.com/

通过

Exchange

(交换器)和

Queues

(队列)发布和消费消息的整个过程是通过一个

Channel

(通道)完成的。

有关路由的详细信息，请访问此链接。

3.2 RabbitMQ 设置

3.2.1 安装

我们可以从这里下载并安装基于我们的操作系统的二进制文件。

然而，在本文中，我们将使用

cloudamqp.com

提供的免费云安装。只需注册服务并登录即可。

在主仪表板上单击

创建新实例

:



然后给你的实例起个名字，然后进入下一步:



然后选择一个可用区:



最后，查看实例信息，点击右下角的

创建实例

:



就是这样。现在在云中运行了一个

RabbitMQ

实例。有关实例的更多信息，请转到您的仪表板并单击

新创建的实例

:



我们可以看到我们可以访问RabbitMQ实例的主机，比如从我们的项目连接所需的用户名和密码:



我们将在Spring应用程序中使用

AMQP URL

连接到这个实例，所以请在某个地方记下它。

您还可以通过单击左上角的

RabbitMQ manager

来查看管理器控制台。这将采用它来管理的您的

RabbitMQ

实例。



Project 配置

现在我们的设置已经准备好了，让我们创建我们的服务:

• cloud-stream-producer-rabbitmq: 作为一个发布者，将消息推送到

  RabbitMQ

• cloud-stream-consumer-rabbitmq: 消费者消费消息

使用

Spring Initializr

producer

REST

选择您喜欢的

Spring Boot

Web

Cloud Stream

Maven

注意：

请注意

cloud-stream

RabbitMQ

Kafka

由于我们将使用

RabbitMQ

Maven

<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

或者，我们也可以将两者结合起来使用

spring-cloud-starter-stream-rabbit

<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

使用同样的方法，创建消费者项目，但仅使用

spring-cloud-starter-stream-rabbit

4. 创建生产者

如前所述，将消息从发布者传递到队列的整个过程是通过通道完成的。因此，让我们创建一个

HelloBinding

greetingChannel

interface HelloBinding {

@Output("greetingChannel")
MessageChannel greeting();
}

因为这将发布消息，所以我们使用

@Output

Channel

现在，让我们创建一个

REST

Channel

@RestController
public class ProducerController {

private MessageChannel greet;

public ProducerController(HelloBinding binding) {
greet = binding.greeting();
}

@GetMapping("/greet/{name}")
public void publish(@PathVariable String name) {
String greeting = "Hello, " + name + "!";
Message<String> msg = MessageBuilder.withPayload(greeting)
.build();
this.greet.send(msg);
}
}

上面，我们创建了一个

ProducerController

MessageChannel

greet

注意: 我们可以用简洁的方式做同样的事情，但是我们使用不同的名称来让您更清楚地了解事物是如何连接的。

然后，我们有一个简单的

REST

PathVariable

name

MessageBuilder

String

MessageChannel

.send()

现在,我们将在的主类中添加

@EnableBinding

HelloBinding

Spring

@EnableBinding(HelloBinding.class)
@SpringBootApplication
public class Application {

public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}

最后，我们必须告诉

Spring

RabbitMQ

AMQP URL

greetingChannel

这两个都是在

application.properties

spring.rabbitmq.addresses=<amqp url>

spring.cloud.stream.bindings.greetingChannel.destination = greetings

server.port=8080

5. 创建消费者

现在，我们需要监听之前创建的通道

greetingChannel

public interface HelloBinding {

String GREETING = "greetingChannel";

@Input(GREETING)
SubscribableChannel greeting();
}

与生产者绑定的两个非常明显区别。因为我们正在消费消息，所以我们使用

SubscribableChannel

@Input

greetingChannel

现在，让我们创建处理数据的方法:

@EnableBinding(HelloBinding.class)
public class HelloListener {

@StreamListener(target = HelloBinding.GREETING)
public void processHelloChannelGreeting(String msg) {
System.out.println(msg);
}
}

在这里，我们创建了一个

HelloListener

processHelloChannelGreeting

@StreamListener

@EnableBinding

HelloBinding

同样，我们在这里使用

@EnableBinding

看看我们的主类，我们没有任何修改:

@SpringBootApplication
public class Application {

public static void main(String[] args) {
SpringApplication.run(Application.class, args);
}
}

在

application.properties

spring.rabbitmq.addresses=<amqp url>
spring.cloud.stream.bindings.greetingChannel.destination=greetings
server.port=9090

6. 全部测试

让我们同时启动生产者和消费者服务。首先，让我们通过点击端点

http://localhost:8080/greet/john

在消费者日志中看到消息内容：

我们使用以下命令启动另一个消费者服务实例(在另一个端口（9091）上):

$ mvn spring-boot:run -Dserver.port=9091

现在，当我们点击生产者

REST

这可能是我们在一些用例中想要的。但是，如果我们只想让一个消费者消费一条消息呢?为此，我们需要在

application.properties

spring.cloud.stream.bindings.greetingChannel.group = greetings-group

现在，再次在不同的端口上运行消费者的2个实例，并通过生产者生产消息再次查看:

这一切也可以在

RabbitMQ

7. 结论

在本文中，我们解释了消息传递的主要概念、它在微服务中的角色以及如何使用

Spring Cloud Stream

RabbitMQ

如Kafka

与往常一样，本文使用的示例代码可以在GitHub获得完整的源代码。

原文：https://stackabuse.com/spring-cloud-stream-with-rabbitmq-message-driven-microservices/

作者：Dhananjay Singh

译者：李东

转载于:https://www.cnblogs.com/liululee/p/11149302.html