在使用spring cloud云架构的时候，我们不得不使用Spring cloud Stream，因为消息中间件的使用在项目中无处不在，我们公司后面做了娱乐方面的APP，在使用spring cloud做架构的时候，其中消息的异步通知，业务的异步处理都需要使用消息中间件机制。spring cloud的官方给出的集成建议（使用rabbit mq和kafka），我看了一下源码和配置，只要把rabbit mq集成，kafka只是换了一个pom配置jar包而已，闲话少说，我们就直接进入配置实施：
1. 简介：
Spring cloud Stream 数据流操作开发包，封装了与Redis,Rabbit、Kafka等发送接收消息。
2. 使用工具：
rabbit，具体的下载和安装细节我这里不做太多讲解，网上的实例太多了
3. 创建commonservice-mq-producer消息的发送者项目，在pom里面配置stream-rabbit的依赖
org.springframework.cloud spring-cloud-starter-stream-rabbit 4. 在yml文件里面配置rabbit mq
server: port: 5666spring: application: name: commonservice-mq-producer profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server # rabbitmq和kafka都有相关配置的默认值，如果修改，可以再次进行配置 stream: bindings: mqScoreOutput: destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghueureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true 5. 定义接口ProducerService
package com.honghu.cloud.producer;import org.springframework.cloud.stream.annotation.Output;import org.springframework.messaging.SubscribableChannel;public interface ProducerService { String SCORE_OUPUT = "mqScoreOutput"; @Output(ProducerService.SCORE_OUPUT) SubscribableChannel sendMessage();} 6. 定义绑定
package com.honghu.cloud.producer;import org.springframework.cloud.stream.annotation.EnableBinding;@EnableBinding(ProducerService.class)public class SendServerConfig {} 7. 定义发送消息业务ProducerController
package com.honghu.cloud.controller;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.integration.support.MessageBuilder;import org.springframework.messaging.Message;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RequestBody;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RequestMethod;import org.springframework.web.bind.annotation.RestController;import com.honghu.cloud.common.code.ResponseCode;import com.honghu.cloud.common.code.ResponseVO;import com.honghu.cloud.entity.User;import com.honghu.cloud.producer.ProducerService;import net.sf.json.JSONObject;@RestController@RequestMapping(value = "producer")public class ProducerController { @Autowired private ProducerService producerService; /** * 通过get方式发送对象 * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/sendObj", method = RequestMethod.GET) public ResponseVO sendObj() { User user = new User(1, "hello User"); Messagemsg = MessageBuilder.withPayload(user).build(); boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); } /** * 通过get方式发送字符串消息 * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/send/{name}", method = RequestMethod.GET) public ResponseVO send(@PathVariable(value = "name", required = true) String name) { Message msg = MessageBuilder.withPayload(name.getBytes()).build(); boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); } /** * 通过post方式发送json对象 * @param name 路径参数 * @return 成功|失败 */ @RequestMapping(value = "/sendJsonObj", method = RequestMethod.POST) public ResponseVO sendJsonObj(@RequestBody JSONObject jsonObj) { Messagemsg = MessageBuilder.withPayload(jsonObj).build(); boolean result = producerService.sendMessage().send(msg); if(result){ return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_SUCCESS, false); } return ResponseCode.buildEnumResponseVO(ResponseCode.RESPONSE_CODE_FAILURE, false); }}
8. 创建commonservice-mq-consumer1消息的消费者项目，在pom里面配置stream-rabbit的依赖
org.springframework.cloud spring-cloud-starter-stream-rabbit
9. 在yml文件中配置：
server: port: 5111spring: application: name: commonservice-mq-consumer1 profiles: active: dev cloud: config: discovery: enabled: true service-id: commonservice-config-server stream: bindings: mqScoreInput: group: honghu_queue destination: honghu_exchange contentType: application/json rabbitmq: host: localhost port: 5672 username: honghu password: honghueureka: client: service-url: defaultZone: http://honghu:123456@localhost:8761/eureka instance: prefer-ip-address: true 10. 定义接口ConsumerService
package com.honghu.cloud.consumer;import org.springframework.cloud.stream.annotation.Input;import org.springframework.messaging.SubscribableChannel;public interface ConsumerService { String SCORE_INPUT = "mqScoreInput"; @Input(ConsumerService.SCORE_INPUT) SubscribableChannel sendMessage();} 11. 定义启动类和消息消费
package com.honghu.cloud;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.netflix.eureka.EnableEurekaClient;import org.springframework.cloud.stream.annotation.EnableBinding;import org.springframework.cloud.stream.annotation.StreamListener;import com.honghu.cloud.consumer.ConsumerService;import com.honghu.cloud.entity.User;@EnableEurekaClient@SpringBootApplication@EnableBinding(ConsumerService.class) //可以绑定多个接口public class ConsumerApplication { public static void main(String[] args) { SpringApplication.run(ConsumerApplication.class, args); } @StreamListener(ConsumerService.SCORE_INPUT) public void onMessage(Object obj) { System.out.println("消费者1，接收到的消息：" + obj); }} 12. 分别启动commonservice-mq-producer、commonservice-mq-consumer1
13. 通过postman来验证消息的发送和接收
可以看到接收到了消息，下一章我们介绍mq的集群方案。
到此，整个消息中心方案集成完毕 完整项目的源码来源 技术支持1791743380
欢迎大家和我一起学习spring cloud构建微服务云架构，我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来，帮助更多有兴趣研发spring cloud框架的朋友，大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。