SpringCloud Netflix复习之Ribbon

文章目录

- 写作背景
- SpringCloud Ribbon是什么，干了什么事情
- Ribbon组件的核心功能
- - Ribbon内置了哪些负载均衡算法
- 上手实战
- - 在SpringCloud里Ribbon实战
- 从源码角度看下Ribbon实现原理
- - SpringCloud与Ribbon整合的原理
  - LoadBalancerInterceptor拦截器改变了RestTemplate什么行为
  - ZoneAwareLoadBalancer默认负载均衡选择server的源码
  - 轮询负载的算法源码
  - 服务名与URI替换并发起Http的源码
  - Ribbon从Eureka Client获取服务列表的源码
  - Ribbon定时更新ServerList源码
  - 自定义更改Ribbon负载规则的原理

写作背景

本文是接上一篇《SpringCloud Netflix复习之Eureka》来写的，有了Eureka之后，微服务之间调用可以通过Eureka Server拉取到服务注册表，就知道了要调用的下游服务的访问IP加端口等信息。但是呢，现在一般微服务都是多实例部署，一来防止单点问题，二来可以水平扩展提高并发能力。那么问题来了，服务A调用服务B，服务B是多实例部署的，我服务A怎么知道要调用服务B哪个实例呢？是随机还是轮询，还是有更智能一些的根据负载和机器配置来选择，Ribbon就是来干这件事的。
本文来复习SpringCloud Ribbon的相关知识，书写思路是以下几个方面

SpringCloud Ribbon是什么，干了什么事情
Ribbon组件的核心功能
上手实战
从源码角度来看下Ribbon的实现原理

SpringCloud Ribbon是什么，干了什么事情

Ribbon本身是Netflix公司(类似国内爱奇艺)研发的一个用于客户端负载均衡的组件，SpringCloud官方拿来封装了一下，成为SpringCloud Netflix生态的一员。刚刚说到了客户端的负载均衡，对应的还有个服务端负载均衡，常见的比如Nginx就是服务端负载均衡，相信你已经懂了一个是从客户端自己发起并执行的，一个是请求到达服务器之后再根据负载算法选择目标服务器。

Ribbon组件的核心功能

Ribbon提供了一系列完善的配置，比如超时配置，重试配置。通过ILoadBalancer获取所有服务实例列表ServerList，然后基于IRule实现的某种负载均衡算法，比如随机、轮询等选出一个服务实例，然后通过RestTemplate发起一个Rest请求。
可以说Ribbon这个中间件最核心的组件就是ILoadBalancer，然后服务实例列表的ServerList，以及用于负载算法IRule从ServerList中选出一个服务实例，还有一个用于ping每个服务实例判断其是否存活的IPing组件。

Ribbon内置了哪些负载均衡算法

Ribbon内置了许多负载均衡算法规则，这些规则的顶级接口是com.netflix.loadbalancer.IRule，可以通过实现IRule接口自定义负载均衡算法，但是一般用内置的负载均衡算法就够了。
在这里插入图片描述

RandomRule：随机找一个服务实例，这种基本不会用。

RoundRobinRule：轮询，从一堆server list中轮询选择出来一个server，每个server平摊到的这个请求，基本上是平均的，比如你100个请求，然后5个实例，基本每个实例会打到20个请求。有个限制默认超过10次获取的server都不可用，会返回空

AvailabilityFilteringRule：顾名思义，这个规则会考察服务器的可用性。如果3次连接失败，就会等待30秒后再次访问；如果某个服务器的连接数超限也就是并发请求太高了，那么会绕过去，不再访问

WeightedResponseTimeRule：带着权重的，每个服务器可以有权重，权重越高优先访问，如果某个服务器响应时间比较长，那么权重就会降低，减少访问

ZoneAvoidanceRule：根据区域来进行负载均衡，说白了就是优先在同一个机房内的节点轮询选取

BestAvailableRule：最小链接数策略，遍历ServerList从中选出一个可用且连接数最小的Server。

RetryRule：重试，默认继承RoundRobinRule就是通过轮询找到的服务器不可用或者请求失败，可以重新找一个服务器，而且没有RoundRoinRule超过10的限制，只要serverList不挂会不停选取判断。

其中ZoneAvoiddanceRule是默认策略，这个在后面源码中也有体现

上手实战

在SpringCloud里Ribbon实战

Springcloud项目中使用Ribbon只需要注入RestTemplate,然后加一个@LoadBalanced注解就可以了

/**
     * Ribbon负载均衡
     *
     * @return RestTemplate
     */
    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        RestTemplate restTemplate = new RestTemplate();
        return restTemplate;
    }

连依赖都不用加，因为Eureka客户端里已经有Ribbon的依赖，所以只要你项目里已经有了eureka客户端的依赖
在这里插入图片描述
RestTemplate本身只是一个Http组件，指定一个url发起一个Http请求，它是不具备负载均衡的功能的，但是这里加了@LoadBalanced注解之后，底层就会用Ribbon实现负载均衡。

我这边演示fc-service-portal(端口是8002)调用fc-service-screen服务，其中fc-service-screen启动两个实例，一个端口是8003，一个端口是8004。
下面是fc-service-portal服务里的Controller代码，通过RestTemplate调用fc-service-screen的/getPort接口，主要是查端口，根据端口号来区分调用的是哪个实例

@RestController
public class HelloWorldController {

    @Resource
    RestTemplate restTemplate;

    @GetMapping("/getPort")
    public int getPort() {
        return restTemplate.getForObject("http://fc-service-screen/getPort", Integer.class);
    }
}

fc-service-sceen8003和8004里Controller里的代码都是下面这样的

@RestController
public class HelloWordController {

    @Value("${server.port}")
    int port;

    @GetMapping("/getPort")
    public int getPort() {
        return port;
    }

}

启动所有服务后，看下Eureka的注册情况
在这里插入图片描述
可以看到fc-service-screen已经有两个实例了
我们请求http://localhost:8002/getPort 看返回结果情况

可以看的出来是轮询的访问方式

从源码角度看下Ribbon实现原理

SpringCloud与Ribbon整合的原理

从@LoadBalanced注解入手


/**
 * Annotation to mark a RestTemplate bean to be configured to use a LoadBalancerClient.
 * @author Spencer Gibb
 */
@Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Qualifier
public @interface LoadBalanced {

}

看注释的意思是@LoadBalanced注解的意思是将一个RestTemplate标记为底层采用LoadBalancerClient来执行实际的Http的请求，支持负载均衡。还是老套路找下XXAutoConfiguration类，找到了LoadBalancerAutoConfiguration

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {
@LoadBalanced
	@Autowired(required = false)
	private List<RestTemplate> restTemplates = Collections.emptyList();

	@Autowired(required = false)
	private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();

	@Bean
	public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(
			final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
		return () -> restTemplateCustomizers.ifAvailable(customizers -> {
			for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
			//定制restTemplate
				for (RestTemplateCustomizer customizer : customizers) {
					customizer.customize(restTemplate);
				}
			}
		});
	}

可以看到有个restTemplate的列表然后每个restTemplate都被定制了，我们找下RestTemplateCustomizer

@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
		final LoadBalancerInterceptor loadBalancerInterceptor) {
		return restTemplate -> {
				List<ClientHttpRequestInterceptor> list = new ArrayList<>(
						restTemplate.getInterceptors());
				list.add(loadBalancerInterceptor);
				//往restTemplate里设置了拦截器LoadBalancerInterceptor
				restTemplate.setInterceptors(list);
		};
}
@Bean
public LoadBalancerInterceptor ribbonInterceptor(
      LoadBalancerClient loadBalancerClient,
      LoadBalancerRequestFactory requestFactory) {
   return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
}

LoadBalancerInterceptor拦截器改变了RestTemplate什么行为

LoadBalancerInterceptor里有个intercept()方法

@Override
	public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
			final ClientHttpRequestExecution execution) throws IOException {
		//获取原始uri信息，参考下面截图
		final URI originalUri = request.getURI();
		//获取服务名就是fc-service-screen
		String serviceName = originalUri.getHost();
		Assert.state(serviceName != null,
				"Request URI does not contain a valid hostname: " + originalUri);
		//拦截请求的服务名，然后构建一个LoadBalancerRequest，然后将服务名和request一起作为参数调用
		//loadBalancer的execute方法，这个方法里会有负载均衡		
		return this.loadBalancer.execute(serviceName,
				this.requestFactory.createRequest(request, body, execution));
	}

originalUri是什么结构
在这里插入图片描述
我们源码跟进去看下

public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
			throws IOException {
		//获取一个LoadBalancer	
		ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
		//负载均衡选取一个server
		Server server = getServer(loadBalancer, hint);
		if (server == null) {
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		RibbonServer ribbonServer = new RibbonServer(serviceId, server,
				isSecure(server, serviceId),
				serverIntrospector(serviceId).getMetadata(server));

		return execute(serviceId, ribbonServer, request);
	}

打个断点看下
在这里插入图片描述
可以看到在SpringCloud整合Ribbon的环境下，使用的默认的LoadBalancer其实就是ZoneAwareLoadBalancer，然后将服务名fc-service-screen根据聚在均衡算法选取一个server，这里面就有和Eureka整合的东西，
通过服务名 ==>ip+端口
将http://fc-service-screen/getPort ==>http://10.100.27.108:8003/getPort
10.100.27.108这个ip就是我本机的ip等同于localhost

ZoneAwareLoadBalancer默认负载均衡选择server的源码

@Override
    public Server chooseServer(Object key) {
        //如果只有一个机房进入这个流程，我本地演示会进入这里
        if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
            logger.debug("Zone aware logic disabled or there is only one zone");
            //调用父类的chooseServer
            return super.chooseServer(key);
        }
        Server server = null;
        try {
            LoadBalancerStats lbStats = getLoadBalancerStats();
            //负载策略是ZoneAvoidanceRule
            Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
            ...省略不用管的代码，多机房一般公司都用不到
    }

单机房逻辑会调用父类也就是BaseLoadBalancer的chooseServer()方法，我打断点进入发现rule其实就是ZoneAvoidanceRule，从源码这里也证明了ZoneAvoidanceRule是默认的负载均衡算法。
在这里插入图片描述
源码继续跟进去，会发现ZoneAvoidanceRule的choose()方法又会调用父类PredicateBasedRule的choose()方法

我们看下获取server的源码实现，AbstractServerPredicate#chooseRoundRobinAfterFiltering

轮询负载的算法源码

private int incrementAndGetModulo(int modulo) {
        for (;;) {
            int current = nextIndex.get();
            int next = (current + 1) % modulo;
            if (nextIndex.compareAndSet(current, next) && current < modulo)
                return current;
        }
    }

这个算法很简单就是简单的轮询
nextIndex是一个AtomicInteger，一开始是0，也就是current是0，然后我们fc-service-screen是两个实例，modulo是2
那么next = (0+1) % 2 = 1,然后设置nextIndex=1 返回的是current=0就是取列表第一个是8003那个实例；
接着第二次请求过来，current就等于1，然后next = (1+1) % 2 = 0，然后设置nextIndex=0，返回current=1就是取列表第二个也就是8004那个实例

服务名与URI替换并发起Http的源码

RibbonLoadBalancerClient#execute

@Override
	public <T> T execute(String serviceId, ServiceInstance serviceInstance,
			LoadBalancerRequest<T> request) throws IOException {
		Server server = null;
		if (serviceInstance instanceof RibbonServer) {
		//这个server已经选好了，比如是10.100.27.108:8004
			server = ((RibbonServer) serviceInstance).getServer();
		}
		if (server == null) {
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		//这个看名字是RibbonLoadBalancer的上下文信息，应该包括了Ribbon的重试配置信息
		RibbonLoadBalancerContext context = this.clientFactory
				.getLoadBalancerContext(serviceId);
		RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server);

		try {
		//看这里调用LoadBalancerRequest的apply()方法
			T returnVal = request.apply(serviceInstance);
			statsRecorder.recordStats(returnVal);
			return returnVal;
		}
		。。。
	}

LoadBalancerRequest是一个匿名内部类，就是在这里定义的

public LoadBalancerRequest<ClientHttpResponse> createRequest(
			final HttpRequest request, final byte[] body,
			final ClientHttpRequestExecution execution) {
		return instance -> {
		//把请求信息封装为一个ServiceRequestWrapper
			HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance,
					this.loadBalancer);
			if (this.transformers != null) {
				for (LoadBalancerRequestTransformer transformer : this.transformers) {
					serviceRequest = transformer.transformRequest(serviceRequest,
							instance);
				}
			}
			//交给ClientHttpRequestExecution去执行Http请求，这里面就是Spring-web包的事情了
			return execution.execute(serviceRequest, body);
		};
	}

ClientHttpRequestExecution肯定会从ServiceRequestWrapper获取请求的URI信息，我们打断点看下
在这里插入图片描述

Ribbon从Eureka Client获取服务列表的源码

Ribbon肯定会和Eureka整合，整合的类EurekaRibbonClientConfiguration，它里面有两个感兴趣的东西

//这个是Ribbon里面关于IPing的
@Bean
@ConditionalOnMissingBean
public IPing ribbonPing(IClientConfig config) {
	if (this.propertiesFactory.isSet(IPing.class, serviceId)) {
		return this.propertiesFactory.get(IPing.class, config, serviceId);
	}
	NIWSDiscoveryPing ping = new NIWSDiscoveryPing();
	ping.initWithNiwsConfig(config);
	return ping;
}
//这个定义了ServerList
@Bean
@ConditionalOnMissingBean
public ServerList<?> ribbonServerList(IClientConfig config,
		Provider<EurekaClient> eurekaClientProvider) {
	if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {
		return this.propertiesFactory.get(ServerList.class, config, serviceId);
	}
	DiscoveryEnabledNIWSServerList discoveryServerList = new DiscoveryEnabledNIWSServerList(
				config, eurekaClientProvider);
	DomainExtractingServerList serverList = new DomainExtractingServerList(
			discoveryServerList, config, this.approximateZoneFromHostname);
	return serverList;
}

在与Eureka整合提供的ServerList里有这个DiscoveryEnabledNIWSServerList类，它里面有个getUpdatedListOfServers()方法，会从Eureka Client获取服务列表

@Override
    public List<DiscoveryEnabledServer> getUpdatedListOfServers(){
    	//看这个方法，里面有一大坨eureka的东西
        return obtainServersViaDiscovery();
    }
    
private List<DiscoveryEnabledServer> obtainServersViaDiscovery() {
        List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>();

        if (eurekaClientProvider == null || eurekaClientProvider.get() == null) {
            logger.warn("EurekaClient has not been initialized yet, returning an empty list");
            return new ArrayList<DiscoveryEnabledServer>();
        }

        EurekaClient eurekaClient = eurekaClientProvider.get();
        if (vipAddresses!=null){
            for (String vipAddress : vipAddresses.split(",")) {
                // if targetRegion is null, it will be interpreted as the same region of client			//看这里
                List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
                for (InstanceInfo ii : listOfInstanceInfo) {
                    if (ii.getStatus().equals(InstanceStatus.UP)) {

                        if(shouldUseOverridePort){
                            if(logger.isDebugEnabled()){
                                logger.debug("Overriding port on client name: " + clientName + " to " + overridePort);
                            }

                            // copy is necessary since the InstanceInfo builder just uses the original reference,
                            // and we don't want to corrupt the global eureka copy of the object which may be
                            // used by other clients in our system
                            InstanceInfo copy = new InstanceInfo(ii);

                            if(isSecure){
                                ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build();
                            }else{
                                ii = new InstanceInfo.Builder(copy).setPort(overridePort).build();
                            }
                        }

                        DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr);
                        serverList.add(des);
                    }
                }
                if (serverList.size()>0 && prioritizeVipAddressBasedServers){
                    break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
                }
            }
        }
        return serverList;
    }

我打断点看下
在这里插入图片描述

初次获取服务列表之后，我们知道Eureka Client默认每30s会从Eureka Server增量拉取注册表并更新本地注册表的，那么Ribbon初次拉取注册表后续如何更新呢？

Ribbon定时更新ServerList源码

大胆猜测一下，Eureka Client是默认每30s更新一次本次注册的表，如果是你来设计Ribbon，你多就更新一次，应该也是30s。Ribbon默认使用的ILoadBalancer是ZoneAwareLoadBalancer，它里面包含IRule、IPing和ServerList组件，我们看下ZoneAwareLoadBalancer的构造方法

public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule,
                                 IPing ping, ServerList<T> serverList, ServerListFilter<T> filter,
                                 ServerListUpdater serverListUpdater) {
        super(clientConfig, rule, ping, serverList, filter, serverListUpdater);
}

调用的是父类DynamicServerListLoadBalancer的构造方法

public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,
                                         ServerList<T> serverList, ServerListFilter<T> filter,
                                         ServerListUpdater serverListUpdater) {
        super(clientConfig, rule, ping);
        this.serverListImpl = serverList;
        this.filter = filter;
        //看这里
        this.serverListUpdater = serverListUpdater;
        if (filter instanceof AbstractServerListFilter) {
            ((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());
        }
        //这个方法里面就有调用从EurekaClient初次拉取注册表以及定时拉取注册表的逻辑
        restOfInit(clientConfig);
    }

serverListUpdater是个啥玩意，我们回到RibbonClientConfiguration类，它里面有注册一个ribbonServerListUpdater，你发现是PollingServerListUpdater

@Bean
@ConditionalOnMissingBean
public ServerListUpdater ribbonServerListUpdater(IClientConfig config) {
	return new PollingServerListUpdater(config);
}

打个断点验证一下
在这里插入图片描述

我们看下restOfInit()方法

void restOfInit(IClientConfig clientConfig) {
        boolean primeConnection = this.isEnablePrimingConnections();
        // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
        this.setEnablePrimingConnections(false);
        //初始化定时拉取任务
        enableAndInitLearnNewServersFeature();
		
        updateListOfServers();
        if (primeConnection && this.getPrimeConnections() != null) {
            this.getPrimeConnections()
                    .primeConnections(getReachableServers());
        }
        this.setEnablePrimingConnections(primeConnection);
        LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
    }

跟进去，看到serverListUpdater.start(updateAction)，这个start一看就很重要

public void enableAndInitLearnNewServersFeature() {
        LOGGER.info("Using serverListUpdater {}", serverListUpdater.getClass().getSimpleName());
        serverListUpdater.start(updateAction);
    }

看下PollingServerListUpdater类的start()方法

@Override
    public synchronized void start(final UpdateAction updateAction) {
        if (isActive.compareAndSet(false, true)) {
            final Runnable wrapperRunnable = new Runnable() {
                @Override
                public void run() {
                    if (!isActive.get()) {
                        if (scheduledFuture != null) {
                            scheduledFuture.cancel(true);
                        }
                        return;
                    }
                    try {
                    //这个doUpdate()就会调DynamicServerListLoadBalancer.this.updateListOfServers()，从Eureka Client拉取注册表
                        updateAction.doUpdate();
                        lastUpdated = System.currentTimeMillis();
                    } catch (Exception e) {
                        logger.warn("Failed one update cycle", e);
                    }
                }
            };
			//开启定时任务了	
            scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                    wrapperRunnable,
                    initialDelayMs,
                    refreshIntervalMs,
                    TimeUnit.MILLISECONDS
            );
        } else {
            logger.info("Already active, no-op");
        }
    }

initialDelayMs和refreshIntervalMs这两个参数没看到地方初始化，直接打断点然后重启fc-service-portal服务，结果发现断点没进来，发起一个请求后进来了，说明Ribbon默认是懒加载的
在这里插入图片描述
可以看到默认延迟1s，然后每30s更新一次注册表
最后再看一眼updateAction.doUpdate();其实它就是实际去拉取注册表的，上面有说到过

public void doUpdate() {
     DynamicServerListLoadBalancer.this.updateListOfServers();
}

自定义更改Ribbon负载规则的原理

我们只需要在配置类里注入IRule这个Bean就可以了，比如我做了如下设置，就是更改负载均衡策略为重试

@Bean
public IRule iRule() {
    //设置重试策略
    return new RetryRule();
}

我们分析一下原理，首先在RibbonClientConfiguration类中有注册IRule为ZoneAvoidanceRule，也就是默认的，它有一个@ConditionalOnMissingBean注解，也就是默认没有IRule的时候用这个

@Bean
@ConditionalOnMissingBean
public IRule ribbonRule(IClientConfig config) {
	if (this.propertiesFactory.isSet(IRule.class, name)) {
		return this.propertiesFactory.get(IRule.class, config, name);
	}
	ZoneAvoidanceRule rule = new ZoneAvoidanceRule();
	rule.initWithNiwsConfig(config);
	return rule;
}

再看ILoadBalancer

@Bean
	@ConditionalOnMissingBean
	public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
			ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
			IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
		if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
			return this.propertiesFactory.get(ILoadBalancer.class, config, name);
		}//rule默认就是ZoneAvoidanceRule
		return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,
				serverListFilter, serverListUpdater);
	}

所以我们配置自己的IRule后，就用我们这个，可以打断点证明一下

@Bean
public IRule iRule() {
    //设置重试策略
    return new RetryRule();
}

在ZoneAwareLoadBalancer父类构造方法里打了断点
在这里插入图片描述

源码可以跟进去看下，在父类BaseLoadBalancer里看看肯定有设置rule的地方

public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {
        initWithConfig(config, rule, ping, createLoadBalancerStatsFromConfig(config));
    }
	//看这个
    void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping) {
        initWithConfig(clientConfig, rule, ping, createLoadBalancerStatsFromConfig(config));
    }

void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping, LoadBalancerStats stats) {
        ...省略
        setRule(rule);
        setPing(ping);

        setLoadBalancerStats(stats);
        rule.setLoadBalancer(this);
       
        。。。省略
        init();

    }