注意：

請注意 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獲得完整的 源代碼 。

原文： stackabuse.com/spring-clou…

作者：Dhananjay Singh

推薦閱讀：