本文节选自《疯狂Spring Cloud微服务架构实战》

京东购买地址：https://item.jd.com/12256011.html

当当网购买地址：http://product.dangdang.com/25201393.html

Spring Cloud教学视频：https://my.oschina.net/JavaLaw/blog/1552993

Spring Cloud电子书：https://my.oschina.net/JavaLaw/blog/1570383

开发消息微服务

        本小节主要讲解如何在微服务中开发消息应用，根据前面章节可知，Spring Cloud Stream帮我们做了一定程度的简化，只需要通过少量的代码配置，就可以实现前面两个框架的功能，而不需要调用它们的API。

准备工作

        我们需要在微服务客户端中实现消息生产者与消费者，先建立以下几个项目：

• 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微服务架构实战》

Spring Cloud教学视频：https://my.oschina.net/JavaLaw/blog/1552993

Spring Cloud电子书：https://my.oschina.net/JavaLaw/blog/1570383

本书代码共享地址：https://gitee.com/yangenxiong/SpringCloud