目录
一、概述
1、简介
2、官网
3、设计思想
二、消息驱动之服务者
1、新建module
2、pom
3、yml
4、主启动
5、业务类
6、测试
三、消息驱动之消费者
1、新建module
2、pom
3、yml
4、主启动
5、业务类
6、测试
四、分组消费与持久化
1、问题
2、解决消息重复消费的问题（分组）
3、通过分组也可解决持久化的问题
五、SpringCloud Sleuth分布式请求链路跟踪
1、概述
2、zipkin
3、搭建服务端的链路配置
4 、搭建客户端的链路配置
5、测试
一、概述
1、简介
屏蔽底层消息中间件的差异,降低切换成本，统一消息的编程模型
官方定义 Spring Cloud Stream 是一个构建消息驱动微服务的框架。 应用程序通过 inputs 或者 outputs 来与 Spring Cloud Stream中binder对象交互。 通过我们配置来binding(绑定) ，而 Spring Cloud Stream 的 binder对象负责与消息中间件交互。 所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。 通过使用Spring Integration来连接消息代理中间件以实现消息事件驱动。 Spring Cloud Stream 为一些供应商的消息中间件产品提供了个性化的自动化配置实现，引用了发布-订阅、消费组、分区的三个核心概念。 目前仅支持RabbitMQ、Kafka。
2、官网
Spring Cloud Streamhttps://spring.io/projects/spring-cloud-stream#overviewSpring Cloud Stream中文指导手册https://m.wang1314.com/doc/webapp/topic/20971999.html
3、设计思想
Stream中的消息通信方式遵循了发布-订阅模式
通常的mq的实现思想
生产者/消费者之间靠消息媒介传递信息内容---Message
消息必须走特定的通道---消息通道MessageChannel
消息通道里的消息如何被消费呢，谁负责收发处理----消息通道MessageChannel的子接口SubscribableChannel，由MessageHandler消息处理器所订阅
Stream为什么可以统一底层差异
在没有绑定器这个概念的情况下，我们的SpringBoot应用要直接与消息中间件进行信息交互的时候，由于各消息中间件构建的初衷不同，它们的实现细节上会有较大的差异性 通过定义绑定器作为中间层，完美地实现了应用程序与消息中间件细节之间的隔离。 通过向应用程序暴露统一的Channel通道，使得应用程序不需要再考虑各种不同的消息中间件实现。通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。
binder
通过定义绑定器Binder作为中间层，实现了应用程序与消息中间件细节之间的隔离。Binder可以生成Binding，Binding用来绑定消息容器的生产者和消费者，它有两种类型，INPUT和OUTPUT，INPUT对应于消费者，OUTPUT对应于生产者。
Stream的工作流程
Binder：很方便的连接中间件，屏蔽差异
Channel：通道，是队列Queue的一种抽象，在消息通讯系统中就是实现存储和转发的媒介，通过Channel对队列进行配置
Source和Sink：简单的可理解为参照对象是Spring Cloud Stream自身， 从Stream发布消息就是输出，接受消息就是输入。
常用注解
二、消息驱动之服务者
1、新建module
cloud-stream-rabbitmq-provider8801
2、pom
3、yml
server:
port: 8801
spring:
application:
name: cloud-stream-provider
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息；
defaultRabbit: # 表示定义的名称，用于于binding整合
type: rabbit # 消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: 192.168.195.132
port: 5672
username: xiaojiang
password: 123
bindings: # 服务的整合处理
output: # 这个名字是一个通道的名称
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://eureka7002.com:7002/eureka/
instance:
lease-renewal-interval-in-seconds: 30 # 设置心跳的时间间隔（默认是30秒）
lease-expiration-duration-in-seconds: 90 # 如果现在超过了5秒的间隔（默认是90秒）
instance-id: send-8801.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址
4、主启动
@SpringBootApplication
@EnableEurekaClient
public class StreamProvider8801 {
public static void main(String[] args) {
SpringApplication.run(StreamProvider8801.class,args);
}
}
5、业务类
服务接口
package com.xiaojiang.service;
public interface IMessageProvider {
public String send();
}
服务接口实现类
package com.xiaojiang.service.Impl;
import com.xiaojiang.service.IMessageProvider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
@Slf4j
@EnableBinding(Source.class)
public class IMessageProviderImpl implements IMessageProvider {
@Resource
private MessageChannel output;
@Override
public String send() {
String serial = UUID.randomUUID().toString();
this.output.send(MessageBuilder.withPayload(serial).build());
log.info(serial);
return serial;
}
}
controller
package com.xiaojiang.controller;
import com.xiaojiang.service.IMessageProvider;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
@RestController
public class SendMessageController {
@Resource
private IMessageProvider provider;
@RequestMapping("/sendmessage")
public String sendMessage(){
return provider.send();
}
}
6、测试
2022-03-21 14:40:40.423 INFO 7028 --- [io-8801-exec-10] c.x.service.Impl.IMessageProviderImpl : d51a2cb2-6b28-4964-96c9-f19d7c7827b9 2022-03-21 14:40:40.630 INFO 7028 --- [nio-8801-exec-8] c.x.service.Impl.IMessageProviderImpl : e63255e5-1c4f-402a-9e89-9fa90cf71b79 2022-03-21 14:40:40.838 INFO 7028 --- [nio-8801-exec-1] c.x.service.Impl.IMessageProviderImpl : e114b904-1044-409e-ba23-fd9366ca80cf
三、消息驱动之消费者
1、新建module
cloud-stream-rabbitmq-consumer8802
2、pom
3、yml
server:
port: 8802
spring:
application:
name: cloud-stream-consumer
cloud:
stream:
binders:
defaultRabbit:
type: rabbit
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: 192.168.195.132
port: 5672
username: xiaojiang
password: 123
bindings:
input:
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client:
service-url:
defaultZone: http://eureka7002.com:7002/eureka/
4、主启动
@SpringBootApplication
@EnableEurekaClient
public class StreamConsumer8802 {
public static void main(String[] args) {
SpringApplication.run(StreamConsumer8802.class,args);
}
}
5、业务类
package com.xiaojiang.controller;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@EnableBinding(Sink.class)
public class ReceiveConsumerListener {
@Value("${server.port}")
String port;
@StreamListener(Sink.INPUT)
public void input(Message
log.info("端口为{}的客户机接收到的消息为{}",port,message.getPayload());
}
}
6、测试
2022-03-21 14:40:40.631 INFO 11984 --- [YW8F4Dnmxv5dA-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为e63255e5-1c4f-402a-9e89-9fa90cf71b79 2022-03-21 14:40:40.840 INFO 11984 --- [YW8F4Dnmxv5dA-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为e114b904-1044-409e-ba23-fd9366ca80cf 2022-03-21 14:40:41.014 INFO 11984 --- [YW8F4Dnmxv5dA-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为b89b8332-f2a6-4b2e-9143-ae7157a1eece
四、分组消费与持久化
按照8802的模板，新建一个8803消费者并运行
1、问题
发送端发送的消息两个消息消费端都收到了，存在消息重复消费的问题
当消息消费者宕机后消息存在丢失问题
2、解决消息重复消费的问题（分组）
方案
分组和持久化属性group
比如在如下场景中，订单系统我们做集群部署，都会从RabbitMQ中获取订单信息， 那如果一个订单同时被两个服务获取到，那么就会造成数据错误，我们得避免这种情况。这时我们就可以使用Stream中的消息分组来解决
注意在Stream中处于同一个group中的多个消费者是竞争关系，就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)， 同一组内会发生竞争关系，只有其中一个可以消费。
将8802和8803变成同一组
group: xiaojiang
8802修改yml
bindings:
input:
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
group: xiaojiang #设置分组
8803修改yml
bindings:
input:
destination: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
group: xiaojiang #设置分组
测试后发现
同一个组的多个微服务实例，每次只会有一个拿到
3、通过分组也可解决持久化的问题
将8802的分组去掉，并将两台消费者宕机发送端发送后消息后先启动8802，获取不到消息启动8803后，获取到了消息
测试结果 2022-03-21 16:05:42.762 INFO 12096 --- [nge.xiaojiang-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为484f2653-baf4-4a40-b657-ce78eb8fdbcf 2022-03-21 16:05:42.763 INFO 12096 --- [nge.xiaojiang-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为9f0b6e0c-7f8e-4b49-a6d2-22f38094fcd5 2022-03-21 16:05:42.765 INFO 12096 --- [nge.xiaojiang-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为ce4d323a-d914-4c14-a542-49bfdd8028ae 2022-03-21 16:05:42.766 INFO 12096 --- [nge.xiaojiang-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为07969516-4179-465a-bdde-e25468ea3e74 2022-03-21 16:05:42.768 INFO 12096 --- [nge.xiaojiang-1] c.x.controller.ReceiveConsumerListener : 端口为8802的客户机接收到的消息为2c75b870-9989-4188-83d3-e7fb7ec9d3c1
五、SpringCloud Sleuth分布式请求链路跟踪
1、概述
在微服务框架中，一个由客户端发起的请求在后端系统中会经过多个不同的的服务节点调用来协同产生最后的请求结果，每一个前段请求都会形成一条复杂的分布式服务调用链路，链路中的任何一环出现高延时或错误都会引起整个请求最后的失败。
Spring Cloud Sleuth提供了一套完整的服务跟踪的解决方案
在分布式系统中提供追踪解决方案并且兼容支持了zipkin
架构图
表示一请求链路，一条链路通过Trace Id唯一标识，Span标识发起的请求信息，各span通过parent id 关联起来
名词解释
Trace:类似于树结构的Span集合，表示一条调用链路，存在唯一标识
span:表示调用链路来源，通俗的理解span就是一次请求信息
2、zipkin
下载
https://sourceforge.net/projects/zipkin.mirror/yxg
运行
在zipkin的jar包所在目录下打开cmd
java -jar zipkin-server-2.12.9-exec.jar
3、搭建服务端的链路配置
用cloud-provider-payment8001作为服务端
pom
添加以下maven坐标
yml添加以下配置
spring:
application:
name: provider-payment
zipkin:
base-url: http://localhost:9411
sleuth:
sampler:
#采样率值介于 0 到 1 之间，1 则表示全部采集
probability: 1
controller
@GetMapping("/zipkin")
public String paymentZipkin()
{
return "hi ,i'am paymentzipkin server fall back，welcome to atguigu，O(∩_∩)O哈哈~";
}
4 、搭建客户端的链路配置
cloud-consumer-payment81作为客户端
pom
yml
spring:
application:
name: comsumer-order
zipkin:
base-url: http://localhost:9411
sleuth:
sampler:
probability: 1
controller
@GetMapping("payment/zipkin")
public String paymentZipkin()
{
String result = restTemplate.getForObject("http://localhost:8001"+"/payment/zipkin/", String.class);
return result;
}
5、测试
浏览器访问 http://localhost:81/comsumer/payment/zipkin
相关阅读
您阅读本篇文章共花了：
发表评论