首页  流程图  详情

微服务源码汇总

2022-08-25 15:06:34   0  举报





nacos-注册中心、配置中心 sentinel

nacos源码

Sentinel源码图解

作者其他创作

大纲/内容

chain.exit

作为初始化@Bean参数

instanceController.beat

cache.md5 = md5; cache.lastModifiedTs = lastModifiedTs; NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey));

通过nacos控制台修改配置发布POST 请求 /v1/cs/configs

SlotChainProvide.newSlotchain()

为每一个dataID注册监听器

逐个调用slot校验链条里的每个校验规则的退出exit逻辑

@PostConstruct

null

2.同步集群其它节点

service.init()

发布ApplicationReadyEvent事件

serviceRegistry.register(this.getRegistration)

springboot 提供的加载配置文件的接口

这个添加的拦截器会拦截所有的WEB请求的URL

如果断路器是半打开状态currentState.get() == State.HALF_OPEN

获取新的参数值，并和之前的比较，找出变化的参数值

调用退出

新增配置

client

本地没取到，远程调用配置中心获取配置信息

SmartApplicationListener

调用异常数及比例判断逻辑

false查本地磁盘文件

拦截把stock-service换成真实的IP

this.registerNacosListenersForApplications()

如果断路器是关闭状态currentState.get() == State.CLOSE

DistroProtocol

1将实例数据更新到内存注册表

定制RestTemplate（将LoadBalancerInterceptor封装到RestTemplate的拦截器集合里去）

statisticSlot.exit

serviceManager.registerInstance

key 判断是否临时数据

抽取SYSTEM，JNDI，SERVLET之外的所有参数变量

NacosAutoServiceRegistration

本节点是否leader

计算当前时间窗口的统计数据

serviceManager.removeInstance

@Bean

spring容器启动refresh()

抛BlockException

抛业务异常Throwable会调用处理注解属性fallback方法

degradeSlot.exit

PushService.onApplicationEvent(ServiceChangeEvent event)

if (instance.isEphemeral()){添加一个延时的心跳任务BeatTask}

Instannce（order）127.0.1.1

Ribbon原理

NacosConfigBootstrapConfiguration

将service的全量实例instance写入内存注册表

DataChangeTask#run

int listenerSize = cacheMap.size(); int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize())

服务端返回接受的心跳信息null，NamingResponseCode.RESOURCE_NOT_FOUND重新注册

logSlot

standalone部署，并且使用的是内置数据库derby

createEmptyService()

增加抛出非限流Block异常调用数

Fegin原理

服务发现

SH-cluster

rt maxAllowedRt

node.increaseThreadNum()

刷新容器中@RefreshScope 标记的类

Instannce（stock）127.0.1.1

handFallBack

InnerWorker.run()

Instannce（pay）127.0.0.1

long currentMaxId = persistService.findConfigMaxId();

DistroConsistencyServiceImpl.init()

first

MetricsHttpAgent#httpGet

HZ-cluster

1.事件是否是ApplicationReadyEvent

是否临时实例ephemeral: true（默认）是存内存false 否存文件

正常情况，则增加当前调用现场数

HttpClient.asyncHttpPostLarge

SentinelWebAutoConfiguration

queue.take()

SpringApplication#run()

集群模式下轮询获取一个服务节点调用

再次比较md5返回

初始化

长轮询

NacosRestTemplate#get

listeners.running(context)

获取配置中心服务ConfigService

通知订阅者

dumpAllProcessor.process(new DumpAllTask())

bind(event)

新节点启动后，从集群其它节点一次性同步数据

SentinelWebAutoConfiguration实现接口WebMvcConfigurer容器启动时会调用其实现方法addInterceptor

NacosConfigService(Properties properties)

sentinel

调用service的实例心跳接口（HttpMethod.POST）

cacheMap放着需要刷新的配置，已3000为一个组

isRefreshEnabled()

fireEntry

源码入口

NacosFactory.createConfigService

实现ApplicationListener接口的类spring容器启动时会调用处理事件的方法

刷新记录

断路器再次打开，按配置的熔断时长更新断路器的熔断周期

资源降级规则

@LoadBalanced

@SentinelResource

prepareEnvironment

isFixedPolling()

chain.entry

SystemSlot

RaftCore

扫描FeginClient注解的类注入spring容器

ServiceManager.init()

Group交易微服务分组

handle(pair)

慢调用比例如果超过配置阈值将断路器打开

NacosRegistration

Instannce（EMS）127.0.1.2

将FeginClient注解的类的信息交给工厂bean代理类并将代理类的定义注册到spring容器

loadExtConfiguration(composite)

spring-cloud-starter-alibaba-nacos-config.jar里的spring.factories文件里

集群节点同步

异步更新内存注册表

调用真实实例接口

DumpService

distroComponentHolder.findDataStorage(type).getDistroData(getDistroKey())

用于分页每次取1000条数据，刷进磁盘和内存

Instannce（order）127.0.0.2

ConfigController#getConfig

对资源逐条校验每个规则

LoadBalancerClient

LoadBalancerInterceptor

hostReactor.getServiceInfo

startDistroTask()

allinstances.addAll(persistenInstances)allinstances.addAll(ephemeralInstances )

nacosConfigManager.getConfigService()

优先使用本地配置

GlobalExecutor.submitLoadDataTask

实现

检查本地配置（容错）

每10s执行一次

canPassCheck

rt = maxAllowedRt

refreshContext(context)

DistroSyncChangeTask.run()

如果超过15s没有收到心跳将Healthy属性设置false

ApplicationListener

notifier.run()

raftStore.write(datum)

加载nacos配置中心的配置信息

扫描EnableFeginClients标签里信息并注册

2.同步实例信息到nacos server 集群其它节点

添加服务健康检测任务

dumpChangeProcessor.process(new DumpChangeTask())

LoadBalancerFeginClient

doSrvIpxt

获取

获取的结果保存到本地

ClientWorker.LongPollingRunnable#run()

Instannce（EMS）127.0.1.1

return

loadNacosDataIfPresent

2.事件是否是RefreshEvent

再尝试调用一次请求

end

对变化的配置，调用监听器处理

NamingService. registerInstance

Instannce（EMS）127.0.0.2

/distro/datum

Service库存服务

NameService. getAllInstances()

断路器关闭currentState.get() == State.CLOSE

同步写实例数据到文件

RestTemplate

extends

cacheData.checkListenerMd5()

对注册的代理工厂bean调用getObject()方法获取真实的代理类

PropertySourceBootstrapConfiguration#initialize

/instance/beat

@AspectSentinelResourceAspect

checkLocalConfig(cacheData)

持久化配置到mysql

Instannce（stock）127.0.0.1

NacosContextRefresher

客户端工作类

延迟29s执行

ApplicationContextInitializer

service.processClientBeat(clientBeat);

ExternalDumpService#init

CtSph.entryWithPirority

将实例数据更新到内存注册表

逐个调用slot校验链条里的每一个校验规则的entry逻辑

加载配置合并到CompositePropertySource

distroProtocol.sync

DistroLoadDataTask.run()

往阻塞队列tasks放入注册的实例数据

重置慢调用和总调用次数为0

调用service的服务发现接口（HttpMethod.GET）传入的参数里有客户端的udp端口，这个是方便服务端实例有变化了通过udp的方式同步给客户端

发布RefreshEvent事件，监听器RefreshEventListener

直接返回客户端NamingResponseCode.RESOURCE_NOT_FOUND

NacosConfigManager#createConfigService

第一次调用服务接口时根据服务名去服务端获取

慢调用比例降级逻辑判断

分页查询

单机限流

获取当前时间的统计时间窗口

implement

implements

增加时间窗口里的请求通过数

AbstractAutoServiceRegistration

/instance

NacosPropertySourceLocator

配置写入磁盘

监听事件变化，回调listener

向nacos服务端发送长连接30s超时返回有变化的dataId

instanceController.register

rt maxAllowedRt如果请求响应时间超过最大rt时间把慢调用次数加1

校验资源系统规则

nacos服务注册中心

利用CountDownLatch实现一个简单的raft协议，必须半数以上的节点写入成功才会返回给客户端

把原来的environment里的参数，放到一个新的Spring context容器下重新加载，完事后关闭新容器，这里就获取参数的新值了

发布环境改变事件，带上变化的参数值

监听ApplicationReadyEvent事件

返回注册时写入得到实例属性

spring容器初始化过程中bean初始化后置处理过程中对AOP的增强处理

SentinelWebInterceptor

FlowSlot.entry

BlockException

ConfigCacheService.dump

NameSpace(Dev)

Instannce（stock）127.0.1.2

DistroTaskEngineHolder

spring-cloud-alibaba-nacos-discovery.jar里的spring.factories文件里EnableAutoConfiguration对应的NacosServiceRegistryAutoConfiguration

putService(service)

NacosConfigProperties

校验资源授权规则

Instannce（pay）127.0.1.2

while (lastMaxId < currentMaxId)

onApplicationEvent

从时间窗口数组里获取依据当前时间计算的下标对应的old时间窗口

定时执行

同步数据到集群其他节点

Group仓库微服务分组

ClientLongPolling clientSub = iter.next()

allSubs.add(this)

如果超过30s没有收到心跳剔除改实例

校验通过，继续触发slot链上的下一个规则

counter.slowCount.add(1)

distroMapper.responsible(serviceName)

DistroController.onSyncDatum

refreshEnvironment()

从内存Map获取当前服务，没有创建

排队等待

SpringApplication#run

用RestTemplateCustomizer定制所有的RestTemplate

ProcessRunnable.run()

nacos更新注册表内存方法里，为了防止读写并发冲突，大量运用copyOnWirte思想。具体做法就是把原内存结构复制一份，操作完最后再替换回真正的注册表内存里去Eureka防止防止读写并发冲突用的是注册表的多级注册缓存结构，只读缓存，读写缓存，内存注册表，各缓存之间定时同步，客户端感知的及时性不及nacos

发布服务变化的事件

GlobalExecutor.submitDistroNotifyTask(notifier)

task.run

NacosServiceRegistry

有一个被@LoadBalanced注解的RestTemplates(装了有一个被所有@LoadBalanced注解的RestTemplate)

初始化的构造函数里会，订阅了LocalDataChangeEvent事件

RestTemplateCustomizer

不通过

否持久话数据阿里CP模式Raft协议

handle((ApplicationReadyEvent) event)

//增加总调用次数slowCount+=counter.slowCount.sum();totalCount+=counter.totalCount.sum();

lookProcessChain

如果old时间窗口为空，则新生成一个时间窗口，并放入时间窗口数组里。

deleteIp

获取资源相关的校验规则

令牌桶算法

span style=\"font-size: inherit;\

Instannce（order）127.0.1.2

DefaultProcessorSlotChain

stock-service2

load()

ConfigController#publishConfig

将新注册实例加入对应服务service的实例列表中去

存在不一样的直接返回

ClientWorker

PropertySourceLocator#locateCollection

pair = tasks.take()

BeatTask

否

int target = distroHash(serviceName) % servers.size()

/v1/cs/listener

nacos Serverstock-service192.168.0.1:8001192.168.0.1:8001

LongPollingService#LongPollingService()

AuthritySlot

distro/datums

addInterceptors

Server

调用service的实例同步接口（HttpMethod.PUT）

线程从tasks里取任务执行

checkFlow

readTimeoutMs 45sheaders参数Long-Pulling-Timeout 30sLong-Pulling-Timeout-No-Hangup true（首次，后面false）

node.addPassRequest(count)

processor.process(task)

curThreadNum.incrementAndGet()

添加监听器

调用Feign的接口通过代理类最终调用LoadBalancerFeginClient的execute方法

校验资源流控规则

是

serviceRegistry是NacosServiceRegistry的实例

如果依据当前时间计算出的时间窗口起始时间大于old时间窗口起始时间则将old时间窗重置，变为当前时间应该落入的时间窗口

NacosRestTemplate#postForm

微服务调用关系

开启一个任务ClientBeatProcess，更新客户端实例最后更新时间

consistencyService.put

ribbonstock-service192.168.0.1:8001192.168.0.1:8001

以Udp方式将服务变动通知发给订阅的客户端

如果实例不存在重新注册（例：网络不通导致实例在服务端被下线或服务端重启临时实例丢失）

SPI加载slot chain加载sentinel-core包下META-INF/services里对应文件的类

ServerStatusReporter.run()

加载顺序：1 文件名（微服务名称）2 文件名.文件扩展名3.文件名-profile.文件扩展名有相同的配置项，后面加载的会覆盖前面加载的

beatReactor.addBeatInfo

NacosConfigAutoConfiguration

serviceProxy. seadBeat(beatInfo)

全量dump配置信息

true查数据库

增加规则校验通过调用数

同步的发起机器就是做健康检查任务的那台机器

和历史对比Md5不相同

NameProxy.syncData

addConfigFilesToEnvironment()

增加当前调用线程数

handFailedTask()

根据变化的dataId向nacos服务端获取配置数据，更新本地快照

加载共享配置文件sharedConfigs

allSubs.remove(ClientLongPolling.this)

NacosConfigManager

客户端长轮询，出现异常，延迟2s执行

RefreshEventListener#onApplicationEvent（）

SystemSlot.entry

滑动时间窗

调用service的实例注册接口（HttpMethod.POST）

配置中心属性配置类，对应bootstrap.yml

获取配置数据

获取当前时间对应的时间窗口

比较md5，不一样的记录在changedGroup

new DistroDelayTaskExecuteEngine()

在spring-cloud-context包中实现，通过SPI机制加载：SpringApplication初始化过程中会通过SpringFactoriesLoader加载ApplicationContextInitializer类型的实现类到initializers中

Service物流服务

注册服务实例信息在集群节点间同步

不成功重试

加载当前应用配置

nodeSelectSlot

SentinelWebInterceptors.preHandle

当前时间对应的时间窗口

调用完处理断路器状态

定时获取客户端最新服务数据更新到本地

是否刷新配置，默认true

Instannce（pay）127.0.1.1

instanceController.list

计算当前时间窗口数组下标

nacos配置中心

Service支付服务

Instannce（EMS）127.0.0.1

构建资源的slot校验链条，每个资源都有自己独立的校验链条有点类似Netty的pipeline

lastMaxId = id > lastMaxId ? id : lastMaxId;

degradeSlot

否则，执行长连接任务

init()

将临时的注册实例更新到cluster的ephemeralInstances 属性上去，服务发现查找临时实例最终从内存里找到的就是这个属性

迭代allSubs队列

registry.addInterceptor

集群节点之间相互同步节点状态。如果有节点宕机了，集群其他节点会感知到并更新集群节点状态，这个会影响心跳任务机器选择的计算

LoadBalance()用jdk动态代理生成代理

内部有两个map，一个clusterMap，一个dataMap。1.取dataMap数据2.拿clusterMap实例健康状态，心跳信息更新实例3.添加新实例

缓存配置信息的Md5到内存中并发布LocalDataChangeEvent事件

statisticSlot

循环从阻塞队列tasks取数据处理

service为null

listener就是Service实例

MetricsHttpAgent#httpPost

this.getRegistration是NacosRegistration实例

OPERATE_LOCK.lock()

springboot 启动时调用PropertySourceLocator.locate(env)用来加载配置信息

LoadBalancerAutoConfiguration

subscriber.onEvent(event)

EventPublishingRunListener#running

start()

PropertySourceLocator

Instannce（pay）127.0.0.2

spring-cloud-start-alibaba-sentinel.jar里的spring.factories文件里SentinelWebAutoConfiguration

NamingProxy.getAllData(targetServer)

entry.exit()

用服务名称hash后对机器数取模，确定执行健康检查的机器

UpdateTask

checkConfigInfo()

counter.totalCount.add(1)

增加被规则限流的调用数

漏桶算法

startLoadTask();

clusterBuilderSlot

快速失败

NacosPropertySourceLocator#locate

拦截

抛Throwable

addConfigInfo

ClientBeatCheckTask

遍历集群中不包括自己的机器，向其发送同步数据的请求，只要有一台获取数据成功，就不再向其它机器请求

getPushService().serviceChanged(this)

instance = serviceManager.getInstance

rollingCounterInSecond.addPass(count)

触发后面的规则校验

Instannce（order）127.0.0.1

update

queue.put(task)

会更新lastRefTime为当前时间

FeignClientFactoryBean(代理Bean)

增量

@AutoConfigurationBefore

DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress())

延时执行的定时任务更新客户端的服务缓存

总调用次数加1

nodeSelectSlot.entry

false（默认）

insert

RibbonAutoConfiguration

时间窗口里的请求数据统计放在一个counter数组里，不同的下标代表不同的数据统计指标

getDistroMapper().responsible(service.getName())

instance为空

ropertyUtil.isDirectRead()

调用service的实例注销接口（HttpMethod.DELETE）

GET请求

stock-service1

发布一个LocalDataChangeEvent事件

生成

查找oldConfigInfo更新当前配置保存旧的配置到历史记录

dumpConfigInfo(dumpAllProcessor)

LongPollingService.ClientLongPolling#run()

1.告诉客户端配置变更

instance.setHealthy(false)

Sphu.entry(resourceName)

clientSub.sendResponse(Arrays.asList(groupKey))

spring 构建bean的过程会执行带有@PostConstruct

ServerStatusReporter.init()

onApplicationEvent(ApplicationReadyEvent event)

加载配置中心配置文件bootstrap.ym

Spring的扩展点之一，在ConfigurableApplication Context#refresh（）之前调用，通常用于需要对应用上下文做初始化的web应用中，比如根据上下文环境注册属性源或激活配置文件等

passLoaclCheck

clientBeat为空

如果总调用次数小于配置最小请求数，则不修改断路器状态，直接返回

继承

遍历查询结果

AuthritySlot.entry

executorService.execute(new LongPollingRunnable(i))

Instannce（stock）127.0.0.2

计算当前时间窗口的起始时间位置

FlowSlot

FeginAutoConfiguration

调用NacosConfigService的构造器

handBlockException

AbstractSharedListener#receiveConfigInfo()

如果old时间窗口起始时间跟依据当前时间计算出的时间窗口起始时间相同，则当前时间应该落在old时间窗内，直接返回old时间窗

lastMaxId=0

serviceProxy. registerService

ephemeralInstances = toUpdateInstances

putServiceAndInit(service)

如果feignClient没有配url

查询mysql获取最大主键值

抛规则校验BlockException会调用处理注解属性blockHandle方法

/v1/cs/configs

ConfigController#listener

心跳周期5s间隔

响应配置发生变化的key

NacosRefreshHistory

Service订单服务

instance.setLastBeat(System.currentTimeMillis())

SlotChainBuilder.build

断路器打开currentState.get() == State.OPEN

/instance/list

listener.receiveConfigInfo(contentTmp)

RibbonRequest

MetricsHttpAgent

写入磁盘

registerFeignClient

最终整合ribbon去实现负载均衡

nacos这种推送模式，对于Zookeeper那种通过tcp长连接来说会节省很多资源，就算大量节点更新也不会让Nacos出现太多的性能瓶颈，在Nacos中客户端如果接受到了udp消息会发送一个Ack，如果一定时间内nacos-server没有收到Ack，那么还会进行重发，当超过一定时间后，就不在重发了，虽然通过udp不能保证真正的送达订阅者，但是Nacos还有定时轮询作为兜底，不需担心数据不会更新的情况。Nacos通过这两种手段，既保证了实时性，有保证了数据更新不会漏掉

ServiceReporter.run()

集群节点状态同步

HealthCheckReactor.scheduleCheck(clientBeatCheckTask)

NacosPropertySourceBuilder#loadNacosData

/raft/datum/commit

statisticSlot.entry

如果超过熔断的时间周期将断路器改为半开状态，尝试着调用一次请求

负责收集资源的路径，并将这些资源的调用路径以树状结构保存起来，用于根据调用路径来限流降级

PropertySourceLoader

SELECT max(id) FROM config_info

增加调用完成数和调用执行执行时间RT减少当前调用线程数

NotifyCenter.publishEvent(event)