这篇文章给大家分享的是有关Spring Cloud如何开发消息微服务的内容。小编觉得挺实用的,因此分享给大家做个参考,一起跟随小编过来看看吧。
我们需要在微服务客户端中实现消息生产者与消费者,先建立以下几个项目:
spring-server:Eureka服务器,端口8761,代码目录“codes\08\8.4\spring-server”。
spring-consumer:消息消费者,Eureka客户端,注册到Eureka,端口8080,代码目录“codes\08\8.4\spring-consumer”。
spring-producer:消息生产者,Eureka客户端,注册到Eureka,端口8081,代码目录“codes\08\8.4\spring-producer”。
整个集群加上消息代理,结构如图8-10所示。
图8-10 程序结构
由于Spring Cloud Stream帮我实现了与消息代者交互的功能,因此对于集群中的生产者与消费者来说,不需要关心外部使用的是哪一个消息框架,本小节的案例,默认使用RabbitMQ,默认情况下,会连接本地的5762端口,如果需要在微服务中修改RabbitMQ的连接信息,可使用以下配置:
spring: application: name: spring-producer rabbitmq: host: localhost port: 5672 username: guest password: guest
在“spring-producer”项目中,加入以下依赖:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-config</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-eureka</artifactId> </dependency> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-rabbit</artifactId> </dependency>
主要引入“spring-coud-starter-stream-rabbit”依赖,该依赖会自动帮我们的项目引入“spring-cloud-stream”以及“spring-cloud-stream-binder”。
接下来,编写发送服务,请见代码清单8-5。
代码清单8-5:codes\08\8.4\spring-producer\src\main\java\org\crazyit\cloud\SendService.java
public interface SendService { @Output("myInput") SubscribableChannel sendOrder(); }
新建一个SendService接口,添加sendOrder方法,该方法使用@Output注解进行修饰。使用该注解,表示会创建“myInput”的消息通道,调用该方法后,会向“myInput”通道投递消息,如果@Output注解不提供参数,则使用方法名作为通道名称。接下来,需要让Spring容器开启绑定的功能,在Application类中,加入@EnableBinding注解,请见代码清单8-6。
代码清单8-6:
codes\08\8.4\spring-producer\src\main\java\org\crazyit\cloud\ProducerApplication.java
@SpringBootApplication @EnableEurekaClient @EnableBinding(SendService.class) public class ProducerApplication { public static void main(String[] args) { SpringApplication.run(ProducerApplication.class, args); } }
@EnableBinding注解中,以SendService.class作为参数,Spring容器在启动时,会自动绑定SendService接口中定义的通道。编写控制器,调用SendService的发送方法,往服务器发送消息,请见代码清单8-7。
代码清单8-7:
codes\08\8.4\spring-producer\src\main\java\org\crazyit\cloud\ProducerController.java
@RestController public class ProducerController { @Autowired SendService sendService; @RequestMapping(value = "/send", method = RequestMethod.GET) public String sendRequest() { // 创建消息 Message msg = MessageBuilder.withPayload("Hello World".getBytes()).build(); // 发送消息 sendService.sendOrder().send(msg); return "SUCCESS"; } }
在控制器中,通过@Autowired自动注入SendService,调用sendOrder方法得到SubscribableChannel(通道)实例,再调用send方法,将“Hello World”字符串发送至“消息代理”中,本例默认的消息代理为RabbitMQ。下面,先实现消息者,再一起整合测试。
消费者项目(spring-consumer)所使用的依赖与生产者一致,先编写接收消息的通道接口,请见代码清单8-8。
代码清单8-8:
codes\08\8.4\spring-consumer\src\main\java\org\crazyit\cloud\ReceiveService.java
public interface ReceiveService { @Input("myInput") SubscribableChannel myInput(); }
在ReceiveService中,定义了一个“myInput”的消息输入通道,接下来,与生产者一样,在启动类中绑定消息通道,请见代码清单8-9。
代码清单8-9:
codes\08\8.4\spring-consumer\src\main\java\org\crazyit\cloud\ReceiverApplication.java
@SpringBootApplication @EnableBinding(ReceiveService.class) public class ReceiverApplication { public static void main(String[] args) { SpringApplication.run(ReceiverApplication.class, args); } @StreamListener("myInput") public void receive(byte[] msg) { System.out.println("接收到的消息: " + new String(msg)); } }
在启动类中,同样使用了@EnableBinding来开启绑定,并声明了通道的接口类。新建了一个receive方法,使用@StreamListener注解进行修饰,声明了订阅“myInput”通道的消息。
依次启动spring-server(8761端口)、spring-producer(8081端口)、spring-consumer(8080端口),访问:http://localhost:8081/send,再打开消息者控制台,可以看到“Hello World”字符串的输出,证明消费者已经可以从消息代理中获取到消息。
前面的例子中,使用了RabbitMQ作为消息代理,如果想使用Kafka,可以更换Maven依赖来实现。在生产者与消费者的pom.xml中,将spring-cloud-starter-stream-rabbit的依赖,修改为spring-cloud-starter-stream-kafka,请见以下配置:
<dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-stream-kafka</artifactId> </dependency>
Spring Cloud提供了绑定器的API,目前实现了RabbitMQ与Kafka的绑定器。在笔者看来,绑定器更像适配器,对于我们的消息程序来说,并不关心使用了哪个消息代理,这些都由绑定器去实现。
感谢各位的阅读!关于“Spring Cloud如何开发消息微服务”这篇文章就分享到这里了,希望以上内容可以对大家有一定的帮助,让大家可以学到更多知识,如果觉得文章不错,可以把它分享出去让更多的人看到吧!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。