整合spring cloud云架构 -消息驱动 Spring Cloud Stream

2022-05-15 13:30:20

在使用spring cloud云架构的时候，我们不得不使用Spring cloud Stream，因为消息中间件的使用在项目中无处不在，我们公司后面做了娱乐方面的APP，在使用spring cloud做架构的时候，其中消息的异步通知，业务的异步处理都需要使用消息中间件机制。spring cloud的官方给出的集成建议（使用rabbit mq和kafka），我看了一下源码和配置，只要把rabbit mq集成，kafka只是换了一个pom配置jar包而已，闲话少说，我们就直接进入配置实施：

简介：

Spring cloud Stream 数据流操作开发包，封装了与Redis,Rabbit、Kafka等发送接收消息。

使用工具：

rabbit，具体的下载和安装细节我这里不做太多讲解，网上的实例太多了

创建commonservice-mq-producer消息的发送者项目，在pom里面配置stream-rabbit的依赖

<groupId>org.springframework.cloud</groupId>  
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>

在yml文件里面配置rabbit mq

Java代码收藏代码
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</span>

eureka:
client:

service-url:  
  defaultZone: http://honghu:123456@localhost:8761/eureka

instance:

prefer-ip-address: true</span>

定义接口ProducerService

Java代码收藏代码
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();

}

定义绑定

Java代码收藏代码
package com.honghu.cloud.producer;

import org.springframework.cloud.stream.annotation.EnableBinding;

@EnableBinding(ProducerService.class)
public class SendServerConfig {

}

定义发送消息业务ProducerController

Java代码收藏代码
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方式发送</span>对象<span style="font-size: 16px;"> 
 * @param name 路径参数 
 * @return 成功|失败 
 */  
@RequestMapping(value = "/sendObj", method = RequestMethod.GET)  
public ResponseVO sendObj() {  
    User user = new User(1, "hello User");  
    <span style="color: #ff0000;">Message<User> msg = MessageBuilder.withPayload(user).build();</span>  
    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方式发送</span>json对象<span style="font-size: 16px;"> 
 * @param name 路径参数 
 * @return 成功|失败 
 */  
@RequestMapping(value = "/sendJsonObj", method = RequestMethod.POST)  
public ResponseVO sendJsonObj(@RequestBody JSONObject jsonObj) {  
    Message<JSONObject> 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);  
}

}

创建commonservice-mq-consumer1消息的消费者项目，在pom里面配置stream-rabbit的依赖

Java代码收藏代码

<groupId>org.springframework.cloud</groupId>  
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>

在yml文件中配置：

Java代码收藏代码
server:
port: 5111
spring:
application:

name: commonservice-mq-consumer1

profiles:

active: dev

cloud:

config:  
  discovery:   
    enabled: true  
    service-id: commonservice-config-server  
      
<span style="color: #ff0000;">stream:  
  bindings:  
    mqScoreInput:  
      group: honghu_queue  
      destination: honghu_exchange  
      contentType: application/json

rabbitmq:

 host: localhost  
 port: 5672  
 username: honghu  
 password: honghu</span>

eureka:
client:

service-url:  
  defaultZone: http://honghu:123456@localhost:8761/eureka

instance:

prefer-ip-address: true

定义接口ConsumerService

Java代码收藏代码
package com.honghu.cloud.consumer;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.messaging.SubscribableChannel;

public interface ConsumerService {

  
<span style="color: #ff0000;">String SCORE_INPUT = "mqScoreInput";  

@Input(ConsumerService.SCORE_INPUT)  
SubscribableChannel sendMessage();</span>

}

定义启动类和消息消费

Java代码收藏代码
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);  
}  
  
<span style="color: #ff0000;">@StreamListener(ConsumerService.SCORE_INPUT)  
public void onMessage(Object obj) {  
    System.out.println("消费者1，接收到的消息：" + obj);  
}</span>

}

分别启动commonservice-mq-producer、commonservice-mq-consumer1
通过postman来验证消息的发送和接收

可以看到接收到了消息，下一章我们介绍mq的集群方案。

到此，整个消息中心方案集成完毕（企业架构源码可以加求球：三五三六二四七二五九）

欢迎大家和我一起学习spring cloud构建微服务云架构，我这边会将近期研发的spring cloud微服务云架构的搭建过程和精髓记录下来，帮助更多有兴趣研发spring cloud框架的朋友，大家来一起探讨spring cloud架构的搭建过程及如何运用于企业项目。

码农公寓

相关文章