1. 程式人生 > >(十七) 整合spring cloud+Springboot雲架構 -訊息驅動 Spring Cloud Stream

(十七) 整合spring cloud+Springboot雲架構 -訊息驅動 Spring Cloud Stream

在使用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的依賴

<span style="font-size: 16px;"><!-- 引入MQ訊息驅動的微服務包,引入stream只需要進行配置化即可,是對rabbit、kafka很好的封裝 -->
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency></span>

4. 在yml檔案裡面配置rabbit mq

<span style="font-size: 16px;">server:
  port: 5666
spring:
  application:
    name: commonservice-mq-producer
  profiles: 
    active: dev
  cloud:
    config:
      discovery: 
        enabled: true
        service-id: commonservice-config-server
  <span style="color: rgb(255, 0, 0);"># 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:
[email protected]
:8761/eureka instance: prefer-ip-address: true</span>

 5. 定義介面ProducerService

<span style="font-size: 16px;">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();
}</span>

6. 定義繫結

<span style="font-size: 16px;">package com.honghu.cloud.producer;

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

@EnableBinding(ProducerService.class)
public class SendServerConfig {

}</span>

7. 定義傳送訊息業務ProducerController

<span style="font-size: 16px;">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: rgb(255, 0, 0);">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);
	}
}
</span>

8. 建立commonservice-mq-consumer1訊息的消費者專案,在pom裡面配置stream-rabbit的依賴

<!-- 引入MQ訊息驅動的微服務包,引入stream只需要進行配置化即可,是對rabbit、kafka很好的封裝 -->
<dependency>
	<groupId>org.springframework.cloud</groupId>
	<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>

9. 在yml檔案中配置:

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: rgb(255, 0, 0);">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:[email protected]: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 {
	
	<span style="color: rgb(255, 0, 0);">String SCORE_INPUT = "mqScoreInput";

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

}

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);
	}
	
	<span style="color: rgb(255, 0, 0);">@StreamListener(ConsumerService.SCORE_INPUT)
	public void onMessage(Object obj) {
		System.out.println("消費者1,接收到的訊息:" + obj);
	}</span>

}

12. 分別啟動commonservice-mq-producer、commonservice-mq-consumer1

13. 通過postman來驗證訊息的傳送和接收



 

 

 

 

 

 

可以看到接收到了訊息,下一章我們介紹mq的叢集方案。

到此,整個訊息中心方案整合完畢(需要原始碼可以直接求求2147775633)

歡迎大家和我一起學習spring cloud構建微服務雲架構,我這邊會將近期研發的spring cloud微服務雲架構的搭建過程和精髓記錄下來,幫助更多有興趣研發spring cloud框架的朋友,大家來一起探討spring cloud架構的搭建過程及如何運用於企業專案。