本文节选自《疯狂Spring Cloud微服务架构实战》
京东购买地址：
当当网购买地址：
Spring Cloud教学视频：
Spring Cloud电子书：
开发消息微服务
本小节主要讲解如何在微服务中开发消息应用，根据前面章节可知，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”项目中，加入以下依赖：
org.springframework.cloud spring-cloud-starter-config org.springframework.cloud spring-cloud-starter-eureka org.springframework.cloud spring-cloud-starter-stream-rabbit
主要引入“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
@RestControllerpublic 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端口），访问：，再打开消息者控制台，可以看到“Hello World”字符串的输出，证明消费者已经可以从消息代理中获取到消息。
更换绑定器
前面的例子中，使用了RabbitMQ作为消息代理，如果想使用Kafka，可以更换Maven依赖来实现。在生产者与消费者的pom.xml中，将spring-cloud-starter-stream-rabbit的依赖，修改为spring-cloud-starter-stream-kafka，请见以下配置：
org.springframework.cloud spring-cloud-starter-stream-kafka
Spring Cloud提供了绑定器的API，目前实现了RabbitMQ与Kafka的绑定器。在笔者看来，绑定器更像适配器，对于我们的消息程序来说，并不关心使用了哪个消息代理，这些都由绑定器去实现。
本文节选自《疯狂Spring Cloud微服务架构实战》
Spring Cloud教学视频：
Spring Cloud电子书：
本书代码共享地址：
