11 zuul路由网关

跳过~过后会发一个脑图 在脑图上可以看到zuul网关

12 Gateway新一代网关

12.1 入门

首先新建一个模块cloud-gateway-gateway9527
pom

<dependencies><!--gateway--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-gateway</artifactId></dependency><!--eureka-client--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><!-- 引入自己定义的api通用包，可以使用Payment支付Entity --><dependency><groupId>com.atguigu.springcloud</groupId><artifactId>cloud-api-commons</artifactId><version>${project.version}</version></dependency><!--一般基础配置类--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

yml

server:port: 9527spring:application:name: cloud-gatewayeureka:instance:hostname: cloud-gateway-serviceclient: #服务提供者provider注册进eureka服务列表内service-url:register-with-eureka: truefetch-registry: truedefaultZone: http://eureka7001.com:7001/eureka

主启动类 需要开启注册到eureka中心的注解

@SpringBootApplication
@EnableEurekaClient
public class GateWayMain9527
{public static void main(String[] args){SpringApplication.run(GateWayMain9527.class,args);}
}

因为这个是一个网关 所以我们不需要业务类
这样一个网关就做好了 9527网关如何做路由映射那？？？
我们目前不想暴露8001端口，希望在8001外面套一层9527

spring:application:name: cloud-gatewaycloud:gateway:routes:- id: payment_routh #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名uri: http://localhost:8001          #匹配后提供服务的路由地址  //即转发地址predicates:- Path=/payment/get/**         # 断言，路径相匹配的进行路由  //即接口- id: payment_routh2 #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名uri: http://localhost:8001          #匹配后提供服务的路由地址predicates:- Path=/payment/lb/**         # 断言，路径相匹配的进行路由

看看controller接口
path填写的就是拦截的方法接口
最后启动
我们在地址栏输入 http://localhost:9527/payment/get/31
网关就会帮我们转发到对应的地址中去 就可以在不 暴露8001接口的情况下 进行数据的访问

当然除了yml方式 我们还可以通过bean的方式进行网关的配置
新建一个配置类：

package com.atguigu.springcloud.config;import org.springframework.cloud.gateway.route.RouteLocator;
import org.springframework.cloud.gateway.route.builder.RouteLocatorBuilder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** @Author XuZhuHong* @CreateTime 2021/11/5 16:42*/
@Configuration
public class GateWayConfig
{/*** 配置了一个id为route-name的路由规则，* 当访问地址 http://localhost:9527/guonei时会自动转发到地址：http://news.baidu.com/guonei* @param builder* @return*/@Beanpublic RouteLocator customRouteLocator(RouteLocatorBuilder builder){RouteLocatorBuilder.Builder routes = builder.routes();routes.route("path_route_atguigu", r -> r.path("/guonei").uri("http://news.baidu.com/guonei")).build();return routes.build();}@Beanpublic RouteLocator customRouteLocator2(RouteLocatorBuilder builder){RouteLocatorBuilder.Builder routes = builder.routes();routes.route("path_route_atguigu2", r -> r.path("/guoji").uri("http://news.baidu.com/guoji")).build();return routes.build();}
}

然后当我们访问 http://localhost:9527/guonei 时就会帮我们 转发到http://news.baidu.com/guonei
但是当我们点击了其他没有做配置的地址时就会报错

12.2 优化

我们发现 上面的方式可以实现路由 但是地址是写死的 当我们服务增多后不方便调用
怎么解决呢？ 通过微服务名实现动态路由
默认情况下Gateway会根据注册中心注册的服务列表，
以注册中心上微服务名为路径创建动态路由进行转发，从而实现动态路由的功能

只需要修改我们路由器的yml即可

server:port: 9527spring:application:name: cloud-gatewaycloud:gateway:discovery:locator:enabled: true #开启从注册中心动态创建路由的功能，利用微服务名进行路由routes:- id: payment_routh #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名# uri: http://localhost:8001          #匹配后提供服务的路由地址uri: lb://cloud-payment-service #匹配后提供服务的路由地址predicates:- Path=/payment/get/**         # 断言，路径相匹配的进行路由- id: payment_routh2 #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名# uri: http://localhost:8001          #匹配后提供服务的路由地址uri: lb://cloud-payment-service #匹配后提供服务的路由地址predicates:- Path=/payment/lb/**         # 断言，路径相匹配的进行路由eureka:instance:hostname: cloud-gateway-serviceclient: #服务提供者provider注册进eureka服务列表内service-url:register-with-eureka: truefetch-registry: truedefaultZone: http://eureka7001.com:7001/eureka

需要注意的是uri的协议为lb，表示启用Gateway的负载均衡功能。

lb://serviceName是spring cloud gateway在微服务中自动为我们创建的负载均衡

12.3 断言

在我们这个配置属性中 我们发现 有一个predicates配置 那么这个是什么呢 就是断言
简单来说就是 条件匹配 匹配的放行 不匹配 拦截
断言有哪些配置的呢？

12.3.1 After 路由断言 Factory

After Route Predicate Factory采用一个参数——日期时间。在该日期时间之后发生的请求都将被匹配。

application.yml

spring:cloud:gateway:routes:- id: after_routeuri: http://example.orgpredicates:- After=2017-01-20T17:42:47.789-07:00[America/Denver]

怎么得到时间串？

ZonedDateTime zbj = ZonedDateTime.now(); // 默认时区System.out.println(zbj);
//        ZonedDateTime zny = ZonedDateTime.now(ZoneId.of("America/New_York")); // 用指定时区获取当前时间
//        System.out.println(zny);

12.3.2 Before 路由断言 Factory

Before Route Predicate Factory采用一个参数——日期时间。在该日期时间之前发生的请求都将被匹配。

application.yml.

spring:cloud:gateway:routes:- id: before_routeuri: http://example.orgpredicates:- Before=2017-01-20T17:42:47.789-07:00[America/Denver]

12.3.3 Between 路由断言 Factory

Between 路由断言 Factory有两个参数，datetime1和datetime2。在datetime1和datetime2之间的请求将被匹配。datetime2参数的实际时间必须在datetime1之后。

application.yml.

spring:cloud:gateway:routes:- id: between_routeuri: http://example.orgpredicates:- Between=2017-01-20T17:42:47.789-07:00[America/Denver], 2017-01-21T17:42:47.789-07:00[America/Denver]

12.3.4 Cookie 路由断言 Factory

Cookie 路由断言 Factory有两个参数，cookie名称和正则表达式。请求包含次cookie名称且正则表达式为真的将会被匹配。

application.yml

spring:cloud:gateway:routes:- id: cookie_routeuri: http://example.orgpredicates:- Cookie=chocolate, ch.p

12.3.5 Header 路由断言 Factory

Header 路由断言 Factory有两个参数，header名称和正则表达式。请求包含次header名称且正则表达式为真的将会被匹配。

application.yml.

spring:cloud:gateway:routes:- id: header_routeuri: http://example.orgpredicates:- Header=X-Request-Id, \d+

12.3.6 Host 路由断言 Factory

Host 路由断言 Factory包括一个参数：host name列表。使用Ant路径匹配规则，.作为分隔符。
application.yml.

spring:cloud:gateway:routes:- id: host_routeuri: http://example.orgpredicates:- Host=**.somehost.org,**.anotherhost.org

12.3.7 Method 路由断言 Factory

Method 路由断言 Factory只包含一个参数: 需要匹配的HTTP请求方式

application.yml.

spring:cloud:gateway:routes:- id: method_routeuri: http://example.orgpredicates:- Method=GET

所有GET请求都将被路由

12.3.8 Path 路由断言 Factory

Path 路由断言 Factory 有2个参数: 一个Spring PathMatcher表达式列表和可选matchOptionalTrailingSeparator标识 .

application.yml.

spring:cloud:gateway:routes:- id: host_routeuri: http://example.orgpredicates:- Path=/foo/{segment},/bar/{segment}

例如: /foo/1 or /foo/bar or /bar/baz的请求都将被匹配

URI 模板变量 (如上例中的 segment ) 将以Map的方式保存于
ServerWebExchange.getAttributes() key为
ServerWebExchangeUtils.URI_TEMPLATE_VARIABLES_ATTRIBUTE.
这些值将在GatewayFilter Factories使用

可以使用以下方法来更方便地访问这些变量。

Map<String, String> uriVariables = ServerWebExchangeUtils.getPathPredicateVariables(exchange);String segment = uriVariables.get("segment");

12.3.9 Query 路由断言 Factory

Query 路由断言 Factory 有2个参数: 必选项 param 和可选项 regexp.

application.yml.

spring:cloud:gateway:routes:- id: query_routeuri: http://example.orgpredicates:- Query=baz

则包含了请求参数 baz的都将被匹配。

application.yml.

spring:cloud:gateway:routes:- id: query_routeuri: http://example.orgpredicates:- Query=foo, ba.

如果请求参数里包含foo参数，并且值匹配为ba. 表达式，则将会被路由，如：bar and baz

12.3.10 RemoteAddr 路由断言 Factory

RemoteAddr 路由断言 Factory的参数为 一个CIDR符号（IPv4或IPv6）字符串的列表，最小值为1，例如192.168.0.1/16（其中192.168.0.1是IP地址并且16是子网掩码）。

application.yml.

spring:cloud:gateway:routes:- id: remoteaddr_routeuri: http://example.orgpredicates:- RemoteAddr=192.168.1.1/24

如果请求的remote address 为 192.168.1.10则将被路由

汇总

 spring:application:name: cloud-gatewaycloud:gateway:discovery:locator:enabled: true #开启从注册中心动态创建路由的功能routes:- id: payment_routh #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名# uri: http://localhost:8001          #匹配后提供服务的路由地址uri: lb://cloud-payment-service #匹配后提供服务的路由地址predicates:- Path=/payment/get/**         # 断言，路径相匹配的进行路由- id: payment_routh2 #payment_route    #路由的ID，没有固定规则但要求唯一，建议配合服务名# uri: http://localhost:8001          #匹配后提供服务的路由地址uri: lb://cloud-payment-service #匹配后提供服务的路由地址predicates:- Path=/payment/lb/**         # 断言，路径相匹配的进行路由- After=2020-02-05T15:10:03.685+08:00[Asia/Shanghai]         # 断言，路径相匹配的进行路由#- Before=2020-02-05T15:10:03.685+08:00[Asia/Shanghai]         # 断言，路径相匹配的进行路由#- Between=2020-02-02T17:45:06.206+08:00[Asia/Shanghai],2020-03-25T18:59:06.206+08:00[Asia/Shanghai]#- Cookie=username,zzyy#- Header=X-Request-Id, \d+  # 请求头要有X-Request-Id属性并且值为整数的正则表达式#- Host=**.atguigu.com- Method=GET- Query=username, \d+  # 要有参数名username并且值还要是整数才能路由

12.4 Filter的使用

路由过滤器可用于修改进入的HTTP请求和返回的HTTP响应，路由过滤器只能指定路由进行使用。

Spring Cloud Gateway 内置了多种路由过滤器，他们都由GatewayFilter的工厂类来产生

如果用官方的 可以参考断言的写法

这里讲一下 自定义全局GlobalFilter 过滤器
自定义过滤器必须实现 两个接口 GlobalFilter,Ordered

package com.atguigu.springcloud.filter;import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import java.util.Date;@Component //必须加，必须加，必须加
public class MyLogGateWayFilter implements GlobalFilter,Ordered
{@Overridepublic Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain){//exchange类似于servlet  //chain 是下一个数据  用于返回 可以看returnSystem.out.println("time:"+new Date()+"\t 执行了自定义的全局过滤器: "+"MyLogGateWayFilter"+"hello");String uname = exchange.getRequest().getQueryParams().getFirst("uname");if (uname == null) {System.out.println("****用户名为null，无法登录");exchange.getResponse().setStatusCode(HttpStatus.NOT_ACCEPTABLE);return exchange.getResponse().setComplete();}return chain.filter(exchange);}@Overridepublic int getOrder(){	//数字越小  执行越靠前return 0;}
}

13 SpringCloud Config分布式配置中心

微服务意味着要将单体应用中的业务拆分成一个个子服务，每个服务的粒度相对较小，因此系统中会出现大量的服务。由于每个服务都需要必要的配置信息才能运行，所以一套集中式的、动态的配置管理设施是必不可少的。

SpringCloud提供了ConfigServer来解决这个问题，我们每一个微服务自己带着一个application.yml，上百个配置文件的管理…
/(ㄒoㄒ)/~~

13.1 使用入门

这里会采用码云的仓库
首先用自己的帐号 创建一个springcloud-config仓库
然后编写几个配置文件
文件名为：

文件内容

config:info: 2021/11/6      config-prod.yml Version=1.0

新建配置中心模块 cloud-config-center-3344

<dependencies><dependency>
<!--            表示这个是配置中心的文件--><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-config-server</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

yml

server:port: 3344spring:application:name:  cloud-config-center #注册进Eureka服务器的微服务名cloud:config:server:git:uri: https://gitee.com/x229827570/springcloud-config/ #仓库上面的git仓库名字####搜索目录search-paths:- springcloud-config####读取分支label: master#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka

主启动类 不要忘记加上注解 @EnableConfigServer 开启配置中心端

@SpringBootApplication
@EnableConfigServer //开启springcloudconfig服务端的注解
public class ConfigCenterMain3344
{public static void main(String[] args) {SpringApplication.run(ConfigCenterMain3344.class, args);}
}

然后运行 当我们访问 http://config-3344.com:3344/master/config-dev.yml 时就会看到如下界面
不要忘记在本地host里做映射哟~

连接成功

13.2 服务端从配置中心获取信息

新建模块cloud-config-client-3355
修改pom

 <dependencies><dependency><!--客户端的config的依赖  和服务端有所不同 --><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-config</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

新建yml
注意 这里的yml有所不同 之前是application.yml
现在需要新建的是bootstrap.yml 那么他是什么？

applicaiton.yml是用户级的资源配置项
bootstrap.yml是系统级的，优先级更加高
Spring Cloud会创建一个“Bootstrap Context”，作为Spring应用的Application Context的父上下文。初始化的时候，Bootstrap Context负责从外部源加载配置属性并解析配置。这两个上下文共享一个从外部获取的Environment。
Bootstrap属性有高优先级，默认情况下，它们不会被本地配置覆盖。 Bootstrap context和Application Context有着不同的约定，所以新增了一个bootstrap.yml文件，保证Bootstrap Context和Application Context配置的分离。
要将Client模块下的application.yml文件改为bootstrap.yml,这是很关键的，
因为bootstrap.yml是比application.yml先加载的。bootstrap.yml优先级高于application.yml

内容

server:port: 3355spring:application:name: config-clientcloud:#Config客户端配置config:label: master #分支名称name: config #配置文件名称profile: dev #读取后缀名称   上述3个综合：master分支上config-dev.yml的配置文件被读取http://config-3344.com:3344/master/config-dev.ymluri: http://localhost:3344 #配置中心地址k#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka

创建启动类

@EnableEurekaClient
@SpringBootApplication
public class ConfigClientMain3355
{public static void main(String[] args){SpringApplication.run(ConfigClientMain3355.class,args);}
}

业务类

@RestController
public class ConfigClientController
{@Value("${config.info}")private String configInfo;@GetMapping("/configInfo")public String getConfigInfo() {return configInfo;}
} 

最后当我们启动时就可以发现一样可以获取文件
但是当我们文件刷新时 发现客户端并不能实时获取到消息
那么怎么办呢

13.3 Config客户端之动态刷新

避免每次更新配置都要重启客户端微服务3355
首先在客户端引入依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

修改yml

# 暴露监控端点
management:endpoints:web:exposure:include: "*" 

再修改controller 在业务类上加上 @RefreshScope 注解

@RestController
@RefreshScope
public class ConfigClientController
{@Value("${config.info}")private String configInfo;@GetMapping("/configInfo")public String getConfigInfo() {return configInfo;}
} 

在维护人员修改配置后 向客户端发送
curl -X POST "http://localhost:3355/actuator/refresh 命令 即可发现已经改变

成功实现了客户端3355刷新到最新配置内容

但是依然存在问题：
怎么解决呢

14 SpringCloud Bus 消息总线

14.1 全局通知

在使用消息总线之前 首先要安装RabbitMQ
这里用的是erlang21.3和rabbitmq-server-3.7.14
然后我们在之前的
cloud-config-center-3344配置中心服务端
cloud-config-client-3355/3366客户端添加RabbitMQ服务支持

<!--添加消息总线RabbitMQ支持-->
<dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

然后在yml配置文件里写对应的配置
首先是3344 对应的配置 两个点

server:port: 3344spring:application:name:  cloud-config-center #注册进Eureka服务器的微服务名cloud:config:server:git:uri: https://gitee.com/x229827570/springcloud-config/ #仓库上面的git仓库名字####搜索目录search-paths:- springcloud-config####读取分支label: master#rabbitmq相关配置   主要是这里------------------rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka##rabbitmq相关配置,暴露bus刷新配置的端点
management:endpoints: #暴露bus刷新配置的端点   主要是这里------------------web:exposure:include: 'bus-refresh'

然后是 3355 和3366端口的yml

server:port: 3366spring:application:name: config-clientcloud:#Config客户端配置config:label: master #分支名称name: config #配置文件名称profile: dev #读取后缀名称   上述3个综合：master分支上config-dev.yml的配置文件被读取uri: http://localhost:3344 #配置中心地址k
#rabbitmq相关配置 15672是Web管理界面的端口；5672是MQ访问的端口
# 主要是这里------------------rabbitmq:host: localhostport: 5672username: guestpassword: guest#服务注册到eureka地址
eureka:client:service-url:defaultZone: http://localhost:7001/eureka
# 暴露监控端点   主要是这里------------------
management:endpoints:web:exposure:include: "*"   # 'refresh'

最后我们先启动eureka和3344 等都启动 完毕 再启动3355 和3366
当我们在码云上修改了代码后 在cmd中执行这个命令 我们就会发现所有的信息都跟着一起修改了

curl -X POST “http://localhost:3344/actuator/bus-refresh”

14.2 定点通知

执行下面代码即可实现

curl -X POST “http://localhost:3344/actuator/bus-refresh/config-client:3355”

bus-refresh就是之前在3344 yml配置中 书写的暴露刷新的接口

15 SpringCloud Stream 消息驱动

15.1 大概讲述

解决了消息插件不同 所带来的问题
屏蔽底层消息中间件的差异,降低切换成本，统一消息的编程模型
为什么使用他

比方说我们用到了RabbitMQ和Kafka，由于这两个消息中间件的架构上的不同，
像RabbitMQ有exchange，kafka有Topic和Partitions分区，

这些中间件的差异性导致我们实际项目开发给我们造成了一定的困扰，我们如果用了两个消息队列的其中一种，后面的业务需求，我想往另外一种消息队列进行迁移，这时候无疑就是一个灾难性的，一大堆东西都要重新推倒重新做，因为它跟我们的系统耦合了，这时候springcloud Stream给我们提供了一种解耦合的方式。

常用的注解如下

15.2 消息的发送操作

在进行这些操作前 我们首先要保证 RabbitMQ环境已经OK
创建一个模块cloud-stream-rabbitmq-provider8801， 作为生产者进行发消息模块
pom

    <dependencies><!--因为这里用的是rabbitmq 所以要引入rabbit的东西--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><!--基础配置--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

yml 他的binder属性可能会报红 不用管 依然可以运行

server:port: 8801spring:application:name: cloud-stream-providercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息；defaultRabbit: # 表示定义的名称，用于于binding整合type: rabbit # 消息组件类型  这里用的是rabbitmqenvironment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服务的整合处理output: # output 代表消息的发送端 这个名字是一个通道的名称 destination: studyExchange # 表示要使用的Exchange名称定义content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka:client: # 客户端进行Eureka注册的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）instance-id: send-8801.com  # 在信息列表时显示主机名称prefer-ip-address: true     # 访问的路径变为IP地址

主启动

@SpringBootApplication
public class StreamMQMain8801
{public static void main(String[] args){SpringApplication.run(StreamMQMain8801.class,args);}
}

一个简单的service接口

public interface IMessageProvider{public String send() ;
}

他的实现类 注意的是 不要引错包了 而且不需要service注解就能实现调用的

package com.atguigu.springcloud.service.impl;import com.atguigu.springcloud.service.IMessageProvider;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;import javax.annotation.Resource;
import java.util.UUID;/*** @Author XuZhuHong* @CreateTime 2021/11/6 18:05*/
@EnableBinding(Source.class) //Source表示是消息的发送者
public class MessageProviderImpl implements IMessageProvider {@Resourceprivate MessageChannel output;  //发送消息需要的组件@Overridepublic String send() {String serial = UUID.randomUUID().toString();//把消息放入消息队列中output.send(MessageBuilder.withPayload(serial).build());System.out.println(serial);return serial;}
}

简单的controller接口

@RestController
public class SendMessageController
{@Resourceprivate IMessageProvider messageProvider;@GetMapping(value = "/sendMessage")public String sendMessage(){return messageProvider.send();}
}

当我们成功跑起这个服务的时候
就可以在robbinMQ的可视化页面中 多了一个通道

这个通道就是我们之前设置的

当我们疯狂刷新的时候 就可以在可视化的界面中 看到波峰图

15.3 消息接收

创建一个 cloud-stream-rabbitmq-consumer8802

   <dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-client</artifactId></dependency><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-stream-rabbit</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-actuator</artifactId></dependency><!--基础配置--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-devtools</artifactId><scope>runtime</scope><optional>true</optional></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency></dependencies>

yml

server:port: 8802spring:application:name: cloud-stream-consumercloud:stream:binders: # 在此处配置要绑定的rabbitmq的服务信息；defaultRabbit: # 表示定义的名称，用于于binding整合type: rabbit # 消息组件类型environment: # 设置rabbitmq的相关的环境配置spring:rabbitmq:host: localhostport: 5672username: guestpassword: guestbindings: # 服务的整合处理input: # input表示他是接收端口 这个名字是一个通道的名称destination: studyExchange # 表示要使用的Exchange名称定义  通道content-type: application/json # 设置消息类型，本次为对象json，如果是文本则设置“text/plain”  接收的数据binder: defaultRabbit # 设置要绑定的消息服务的具体设置eureka:client: # 客户端进行Eureka注册的配置service-url:defaultZone: http://localhost:7001/eurekainstance:lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）instance-id: receive-8802.com  # 在信息列表时显示主机名称prefer-ip-address: true     # 访问的路径变为IP地址

主启动类

@SpringBootApplication
public class StreamMQMain8802{public static void main(String[] args)    {SpringApplication.run(StreamMQMain8802.class,args);}
}

controller类 需要加上@EnableBinding(Sink.class) 注解 表示开启 并且是接收方

@Controller
@EnableBinding(Sink.class) //EnableBinding开启这个消息绑定  sink表示这个是接收方
public class ReceiveMessageListener {@Value("${server.port}")private String serverPort;@StreamListener(Sink.INPUT)  //Sink.INPUT表示这是接收的方法//简单的一个接收的方法  其中的接收的方法参数为固定的public void input(Message<String> message){System.out.println("消费者1号，------->接收到的消息：" + message.getPayload()+"\t port: "+serverPort);}
}

那么当我们把这个类启动
再访问 消息发送者 http://localhost:8801/sendMessage
就会在我们接受者方看到消息

15.4分组消费与持久化

当我们再仿造8802端口8803 运行 我们发现 发送消息的时候 两个消费者都会读取这条消息
造成重复消费 怎么解决呢

微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。
不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。

其实很简单加上group即可 两个group相同 就可以解决这个问题了

另外添加分组还可以有效的防止消息丢失 （即当消息发送者发送了消息 然后没有接收者 当我们启动 没有分组的接收者时 就不会接收到消息 但是启动分了组的接收者就可以接收 ）