本篇内容主要讲解“消息驱动Spring Cloud Stream怎么配置”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“消息驱动Spring Cloud Stream怎么配置”吧!
专注于为中小企业提供网站设计、成都做网站服务,电脑端+手机端+微信端的三站合一,更高效的管理,为中小企业从江免费做网站提供优质的服务。我们立足成都，凝聚了一批互联网行业人才，有力地推动了上1000+企业的稳健成长，帮助中小企业通过网站建设实现规模扩充和转变。
在使用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: 5666 spring:
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: honghu
eureka:
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");
Message msg = 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) {
Message msg = 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: 5111 spring:
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: honghu
eureka:
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来验证消息的发送和接收
到此，相信大家对“消息驱动Spring Cloud Stream怎么配置”有了更深的了解，不妨来实际操作一番吧！这里是创新互联网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！
网页题目：消息驱动SpringCloudStream怎么配置
文章路径：
http://cdkjz.cn/article/gjjoio.html
