注意：

请注意 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

推荐阅读：