用户注册



邮箱:

密码:

用户登录


邮箱:

密码:
记住登录一个月忘记密码?

发表随想


还能输入:200字

cloud云    -  云代码空间

——

(十七) 整合spring cloud云架构 -消息驱动 Spring Cloud Stream

2018-08-30|554阅||

摘要:(十七) 整合spring cloud云架构 -消息驱动 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: #ff0000;"># 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>

 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: #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);
	}
}
</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: #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
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: #ff0000;">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: #ff0000;">@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的集群方案。

到此,整个消息中心方案集成完毕 完整项目的源码来源 技术支持yuncode.net

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


顶 2踩 0收藏
文章评论
    发表评论

    个人资料

    • 昵称: cloud云
    • 等级: 高级设计师
    • 积分: 5150
    • 代码: 0 个
    • 文章: 216 篇
    • 随想: 0 条
    • 访问: 11 次
    • 关注

    人气代码

      标签

      java(57)api(3)教程(27)Spring(60)cache(2)+(8)springmvc(3)mybatis(4)(十)(2)(十一)(2)Cloud(29)Cloud-hong(1)Cloud分布式微服(1)Cloud-Hong(1)Cloud分布式微服(1)Cloud-鸿鹄Cl(1)Cloud--Hon(1)Cloud分布式微服(1)Cloud--Hon(1)Cloud分布式微服(1)Cloud--hon(1)Cloud分布式微服(1)Cloud--Hon(1)分布式微服务云系统-(1)企业分布式微服务云架(1)Cloud-Hong(1)Cloud分布式微服(1)Boot(4)企业分布式微服务云架(1)Cloud--hon(1)Cloud分布式微服(1)Cloud--Hon(1)分布式微服务云系统-(1)Cloud--Hon(1)Cloud分布式微服(1)企业分布式微服务云架(1)Cloud--hon(1)Cloud分布式微服(1)Cloud--Hon(1)分布式微服务云系统-(1)cloud--鸿鹄C(1)源码(12)-(22)cloud集成项目(1)Cloud云服务(1)整合spring(12)cloud云服务架构(5)介绍一下Spring(1)Cloud微服务架构(1)总结Spring(1)Cloud云架构(14)登出流程(3)(2)Cloud+Spri(1)Boot+Mybat(1)+微服务(1)企业分布式微服务云架(1)Spring+(1)SSO单点登录应用认(1)跟大家介绍一下关于S(1)Consul(1)Cloud构建微服务(1)Cloud构建微服务(1)Cloud构建微服务(1)SSO单点登录之OA(1)根据token获取用(1)Netflix(3)springclou(1)RestTempla(1)HongHu云架构c(1)Commons(1)普通抽象(1)Config-Git(1)Cloud--Hon(1)分布式微服务云系统-(1)Cloud微服务分布(1)企业分布式微服务云架(1)Cloud自定义引导(1)SSO单点登录之OA(1)SSO单点登录之OA(1)SSO单点登录之OA(1)cloud云架构-S(1)企业分布式微服务云架(1)企业分布式微服务云架(1)Cloud--Hon(1)Cloud分布式微服(1)项目过程构建(1)分享Spring(1)Config服务器(1)Cloud微服务架构(1)Cloud微服务分布(1)Cloud-hong(1)Cloud分布式微服(1)springclou(1)Cloud微服务分布(1)SpringBoot(59)Cloud各个组件配(1)RestTempla(1)Cloud分布式微服(1)Cloud自定义引导(1)Cloud构建微服务(1)Cloud构建微服务(1)(十七)(1)-消息驱动(1)(十六)(1)-使用spring(1)企业级(29)(二)(3)cloud项目(1)(十二)(2)(九)整合sprin(1)commonserv(1)SSO单点登录之OA(1)SSO单点登录之OA(1)SSO单点登录之OA(1)(九)整合sprin(1)commonserv(1)SSO单点登录之OA(1)SSO单点登录之OA(1)SSO单点登录之OA(1)(十三)(1)SSO单点登录之OA(1)B2B2C(53)(九)Java(1)o2o多用户商城(24)(十)Java(1)(十一)Java(2)(十二)(1)springclou(1)springclou(1)springclou(1)SSO单点登录之OA(1)springclou(1)SSO单点登录之OA(1)springclou(1)(十三)java(2)b2b2c多用户商城(7)(十四)Java(1)(十五)Java(2)(十六)Java(2)b2b2c多用户商城(1)SSO单点登录之OA(1)springclou(1)springclou(1)springclou(1)springclou(1)springclou(1)springclou(1)(十七)Java(1)springclou(1)springclou(1)项目构建过程(七)(2)mybatis电子商(1)Springclou(1)Cloud构建分布式(1)SpringClou(1)Config)(1)ssm(15)springclou(1)b2b2c多用户商城(1)B2B2C多租户电子(1)springclou(1)springclou(1)springclou(1)springclou(1)企业级java(5)shop(6)多用户商城系统源码(2)(十二)java(1)关于SpringCl(1)Bus)(1)b2bc商城系统开源(1)Sleuth)(1)多用户商城系统源码-(1)微服务java_b2(1)springclou(1)springclou(1)springclou(1)springclou(1)b2bc商城系统开源(1)配置自动刷新(1)(四):(1)b2b2c多用户商城(1)Dashboard和(1)b2b2c多用户商城(1)b2b2c多用户商城(1)多用户商城系统(1)(十)用spring(2)springboot(1)(十三)spring(1)微服务java_b2(1)(十二)spring(1)mybatis多租户(1)多级分销spring(1)mybatis多租户(1)多级分销spring(1)mybatis多租户(1)b2b2c多用户商城(1)(二)Spring(3)Boot配置文件详解(1)springclou(1)(一)构建第一个Sp(1)多用户商城系统(三)(1)微服务java_b2(1)整合JPA(1)springclou(1)多级分销Spring(1)多级分销spring(1)mybatis多租户(1)多级分销spring(1)mybatis多租户(1)mybatis多租户(1)多级分销Spring(1)cloud微服务分布(1)Cloud集成项目简(1)cloud微服务分布(1)整合企业架构的技术点(1)(一)spring(1)(十三)java(1)(五)JAVA(1)多用户商城系统源码:(1)b2b2c多用户商城(1)springclou(1)(十二)Spring(1)多用户商城系统:使用(1)Sleuth和Zip(1)(十)java(2)多用户商城系统源码:(1)(十一)企业级jav(1)b2bc商城系统开源(1)(八)java(1)(九)Java(1)多级分销spring(1)mybatis多租户(1)springclou(1)springclou(1)b2b2c多用户商城(1)(十二)微服务多用户(1)springclou(1)Springclou(1)分布式环境下自动发现(1)springclou(1)springclou(1)b2b2c多用户商城(1)b2bc商城系统开源(1)多级分销Spring(1)Dashboard和(1)多级分销Spring(1)多级分销spring(1)mybatis多租户(1)springclou(1)(十一)Java(1)springclou(1)b2bc商城系统开源(1)springclou(1)b2b2c多用户商城(1)多用户商城系统源码((1)springboot(1)Java多用户商城系(1)多用户商城系统(一)(1)springclou(1)Springboot(1)springclou(1)b2b2c多用户商城(1)b2b2c多用户商城(1)springboot(1)(十四)(1)多级Springbo(1)Springboot(1)b2b2c多用户商城(1)springclou(1)springclou(1)springclou(1)springclou(1)common-ser(1)b2b2c多用户商城(1)b2b2c多用户商城(1)企业云架构commo(1)b2bc商城系统开源(1)common-ser(1)微服务java_b2(1)多租户电子商城系统(1)b2b2c多用户商城(1)多级分销Spring(1)整合企业架构的技术点(1)springboot(1)企业云架构commo(1)微服务架构(2)(七)springb(1)(八)springb(1)(九)springb(1)Restdocs创建(1)(十一)spring(1)(十二)spring(1)(十四)在sprin(1)(十五)Spring(1)(十六)用restT(1)(十六)用restT(1)cloud微服务分布(1)SpringClou(1)服务的注册与发现(E(1)SpringClou(1)服务消费者(rest(1)SpringClou(1)服务消费者(rest(1)SpringClou(1)关于SpringCl(1)springclou(1)Cloud微服务架构(1)(六)Spring(1)Cloud云架构一服(1)(三)Spring(1)Cloud云架构一定(1)互联网分布式微服务云(1)(十六)用restT(1)教程(十五)Spri(1)(十四)在sprin(1)Config-Git(1)springclou(1)commonserv(1)commonserv(1)SSO单点登录之OA(1)Cloud云架构-S(1)根据token获取用(1)SpringClou(1)SpringClou(1)企业分布式微服务云S(1)(四)断路器(Hys(1)企业分布式微服务云S(1)(五)路由网关(zu(1)企业分布式微服务云S(1)(六)分布式配置中心(1)企业分布式微服务云S(1)(七)高可用的分布式(1)SpringClou(1)服务消费者(rest(1)(十七)上传文件(1)Tasks)(1)(七)springb(1)(八)springb(1)(九)springb(1)Restdocs创建(1)(十一)spring(1)(十二)spring(1)(十三)spring(1)(十四)在sprin(1)(十五)Spring(1)(十六)用restT(1)(十八)定时任务(S(1)

      最新提问

        站长推荐