您现在的位置是:首页 > 文章详情

【本人秃顶程序员】使用Spring Cloud Stream和RabbitMQ实现事件驱动的微服务

日期:2019-01-29点击:315

←←←←←←←←←←←← 快!点关注

让我们展示如何使用Spring Cloud Stream来设计事件驱动的微服务。首先,Spring Cloud Stream首先有什么好处?因为Spring AMPQ提供了访问AMPQ工件所需的一切。如果您不熟悉Spring AMPQ,请查看此repo,其中包含许多有用的示例。那么为什么要使用Spring Cloud Stream ......?

Spring Cloud Stream概念

  • Spring Cloud Stream通过Binder概念将使用过的消息代理与Spring Integration消息通道绑定在一起。支持RabbitMQ和Kafka。
  • Spring Cloud Stream将基础架构配置从代码中分离为属性文件。这意味着即使您更改了底层代理,您的Spring Integration代码也将是相同的!

示例中的Spring Cloud Stream概念(RabbitMQ)

让我们有一个名为streamInput的交换,它有两个队列streamInput.cities和streamInput.persons。现在让我们将这两个队列插入两个消息通道citiesChannel和personsChannel来消费来自它的传入消息。使用Spring AMPQ,您需要创建SimpleMessageListenerContainer并在代码中连接基础结构。但这有很多样板代码。使用Spring Cloud Stream,您可以将AMPQ配置分离到属性文件:

spring.cloud.stream.bindings.citiesChannel.destination=streamInput spring.cloud.stream.bindings.citiesChannel.group=cities spring.cloud.stream.rabbit.bindings.citiesChannel.consumer.durableSubscription=true spring.cloud.stream.rabbit.bindings.citiesChannel.consumer.bindingRoutingKey=cities spring.cloud.stream.bindings.personsChannel.destination=streamInput spring.cloud.stream.bindings.personsChannel.group=persons spring.cloud.stream.rabbit.bindings.personsChannel.consumer.durableSubscription=true spring.cloud.stream.rabbit.bindings.personsChannel.consumer.bindingRoutingKey=persons 

配置详细信息

在类路径上使用RabbitMQ Binder,每个目标都映射到TopicExchange。在示例中,我创建了名为streamInput的TopicExchange, 并将其附加到两个消息通道citiesChannel和personsChannel。

spring.cloud.stream.bindings.citiesChannel.destination = streamInput  spring.cloud.stream.bindings.personsChannel.destination = streamInput 

现在您需要了解RabbitMQ绑定器的灵感来自Kafka,队列的消费者被分组到消费者组中,其中只有一个消费者将获得消息。这是有道理的,因为您可以轻松扩展消费者。

因此,让我们创建两个队列streamInput.persons和streamInput.cities并将它们附加到streamInput TopicExchange和提到的消息通道

# This will create queue "streamInput.cities" connected to message channel citiesChannel where input messages will land. spring.cloud.stream.bindings.citiesChannel.group=cities  # Durable subscription, of course. spring.cloud.stream.rabbit.bindings.citiesChannel.consumer.durableSubscription=true  # AMPQ binding to exchange (previous spring.cloud.stream.bindings.<channel name>.destination settings). # Only messages with routingKey = 'cities' will land here. spring.cloud.stream.rabbit.bindings.citiesChannel.consumer.bindingRoutingKey=cities  spring.cloud.stream.bindings.personsChannel.group=persons spring.cloud.stream.rabbit.bindings.personsChannel.consumer.durableSubscription=true spring.cloud.stream.rabbit.bindings.personsChannel.consumer.bindingRoutingKey=persons 

连接属性到Spring Integration

好的,到目前为止我创建了两个队列。StreamInput.cities绑定到citiesChannel。StreamInput.persons绑定到peopleChannel。

<destination>.<group>是Spring Cloud Stream约定的队列命名,现在让我们将它连接到Spring Integration:

package com.example.spring.cloud.configuration; import org.springframework.cloud.stream.annotation.Input; import org.springframework.messaging.SubscribableChannel; /**  \* Created by tomask79 on 30.03.17.  */ public interface SinkRabbitAPI {     String INPUT_CITIES = "citiesChannel";     String INPUT_PERSONS = "personsChannel";     @Input(SinkRabbitAPI.INPUT_CITIES)     SubscribableChannel citiesChannel();     @Input(SinkRabbitAPI.INPUT_PERSONS)     SubscribableChannel personsChannel(); } 

Spring Boot启动时加载这个属性

package com.example.spring.cloud; import com.example.spring.cloud.configuration.SinkRabbitAPI; import com.example.spring.cloud.configuration.SourceRabbitAPI; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableBinding({SinkRabbitAPI.class}) public class StreamingApplication {     public static void main(String\[\] args) {         SpringApplication.run(StreamingApplication.class, args);     } } 

在此之后,我们可以创建消费者从绑定的消息通道中的队列接收消息:

import com.example.spring.cloud.configuration.SinkRabbitAPI; import com.example.spring.cloud.configuration.SourceRabbitAPI; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.stereotype.Service; import javax.annotation.Resource; /**  \* Created by tomask79 on 30.03.17.  */ @Service public class ProcessingAMPQEndpoint {     @StreamListener(SinkRabbitAPI.INPUT_CITIES)     public void processCity(final String city) {         System.out.println("Trying to process input city: "+city);     }     @StreamListener(SinkRabbitAPI.INPUT_PERSONS)     public void processPersons(final String person) {         System.out.println("Trying to process input person: "+person);     } } 

RabbitMQ绑定器和代理配置

Spring Cloud Stream如何知道在哪里寻找消息中间件?如果在类路径中找到RabbitMQ绑定器,则使用默认RabbitMQ主机(localhost)和端口(5672)连接到RabbitMQ服务器。如果您的消息中间件配置在不同端口,则需要配置属性:

spring:   cloud:     stream:       bindings:         ...       binders:           rabbitbinder:             type: rabbit             environment:               spring:                 rabbitmq:                   host: rabbitmq                   port: 5672                   username: XXX                   password: XXX 

测试消息消费

  • 安装并运行RabbitMQ代理
  • rabbitmq.git
  • mvn clean install
  • java -jar target / streaming-0.0.1-SNAPSHOT.jar
  • 现在使用路由键'cities'或'persons'在streamInput Exchange上发布消息...输出应该是:
Started StreamingApplication in 6.513 seconds (JVM running for 6.92)  Trying to process input city: sdjfjljksdflkjsdflkjsdfsfd Trying to process input person: sdjfjljksdflkjsdflkjsdfsfd 

使用Spring Cloud Stream重新传递消息

您通常希望在进入DLX交换之前再次尝试接收消息。首先,让我们配置Spring Cloud Stream尝试重新发送失败消息的次数:

spring.cloud.stream.bindings.personsChannel.consumer.maxAttempts = 6 

这意味着如果从streamInput.persons队列接收的消息出错,那么Spring Cloud Stream将尝试重新发送六次。让我们试试,首先让我们修改接收端点以模拟接收崩溃:

 @StreamListener(SinkRabbitAPI.INPUT_PERSONS)     public void processPersons(final String person) {         System.out.println("Trying to process input person: "+person);         throw new RuntimeException();     } 

如果我现在尝试使用人员路由键将某些内容发布到streamInput交换中,那么这应该是输出:

Trying to process input person: sfsdfsdfsd Trying to process input person: sfsdfsdfsd Trying to process input person: sfsdfsdfsd Trying to process input person: sfsdfsdfsd Trying to process input person: sfsdfsdfsd Trying to process input person: sfsdfsdfsd  Retry Policy Exhausted         at org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer.recover (RejectAndDontRequeueRecoverer.java:45) ~\[spring-rabbit-1.7.0.RELEASE.jar! /:na\]         at org.springframework.amqp.rabbit.config.StatelessRetryOperationsInterc        

建议将Spring Cloud Stream 用于事件驱动的MicroServices,因为它可以节省时间,而且您不需要为Java中的AMPQ基础架构编写样板代码。

写在最后:

秃顶程序员的不易,看到这里,点了关注吧!
点关注,不迷路,持续更新!!!

原文链接:https://my.oschina.net/u/4065580/blog/3007153
关注公众号

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。

持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

转载内容版权归作者及来源网站所有,本站原创内容转载请注明来源。

文章评论

共有0条评论来说两句吧...

文章二维码

扫描即可查看该文章

点击排行

推荐阅读

最新文章