2.SpringCloud学习(二)——Spring Cloud Eureka 服务注册中心
2024-05-12 13:27:02
1.简介
1.1 概述
Service Discovery is one of the key tenets of a microservice-based architecture. Trying to hand-configure each client or some form of convention can be difficult to do and can be brittle. Eureka is the Netflix Service Discovery Server and Client. The server can be configured and deployed to be highly available, with each server replicating state about the registered services to the others.
服务发现是基于微服务的体系结构的主要宗旨之一。尝试手动配置每个客户端或某种形式的约定可能很困难并且很脆弱。 Eureka是Netflix Service Discovery服务器和客户端。可以将服务器配置和部署为高度可用，每个服务器将有关已注册服务的状态复制到其他服务器。
1.2 特点
The Eureka server does not have a back end store, but the service instances in the registry all have to send heartbeats to keep their registrations up to date (so this can be done in memory). Clients also have an in-memory cache of Eureka registrations (so they do not have to go to the registry for every request to a service).
By default, every Eureka server is also a Eureka client and requires (at least one) service URL to locate a peer. If you do not provide it, the service runs and works, but it fills your logs with a lot of noise about not being able to register with the peer.
Eureka服务器没有后端存储，但是注册表中的所有服务实例都必须发送心跳信号以使其注册保持最新（这样可以在内存中完成）。客户端还具有Eureka注册的内存缓存（因此，不用每个请求都转发到注册中心）。
默认情况下，每个Eureka服务器也是Eureka客户端，并且需要（至少一个）服务URL来定位对等方。如果您不提供该服务，则该服务将运行并工作，但是它将使您的日志充满无法注册到对等方的噪音。
2.演示环境
- JDK 1.8.0_201
- Spring Boot 2.2.0.RELEASE、Spring Cloud Hoxton.RELEASE
- 构建工具（apache maven 3.6.3）
- 开发工具（IntelliJ IDEA ）
3.演示代码
- nfx-eureka-client： eureka 客户端，注册到 eureka 服务端：
- user-api： 公共api，定义实体和接口；
- user-service-provider： 服务提供方，注册到 eureka server
- user-service-consumer： 服务调用方，注册到 eureka server
- nfx-eureka-server： eureka 服务端，负责提供服务注册及发现功能。
3.1 nfx-eureka-server
3.1.1 代码说明
eureka 服务端，供客户端进行注册，同时提供服务发现功能。
3.1.2 maven 依赖
<dependencies><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-netflix-eureka-server</artifactId></dependency>
</dependencies>
3.1.3 配置文件
spring.application.name=nfx-eureka-server
# 端口号
server.port=9090
# 服务注册中心主机名
eureka.instance.hostname=localhost
# 是否注册自己
eureka.client.register-with-eureka=false
# 是否检索服务
eureka.client.fetch-registry=false
# eureka server 地址
eureka.client.service-url.defaultZone=http://${eureka.instance.hostname}:${server.port}/eureka/
3.1.4 java代码
NetflixEurekaServerApplication.java
// 通过 @EnableEurekaServer 声明为 eureka 服务端
@EnableEurekaServer
@SpringBootApplication
public class NetflixEurekaServerApplication {public static void main(String[] args) {SpringApplication.run(NetflixEurekaServerApplication.class, args);}
}
3.2 nfx-eureka-client
3.2.1 user-api
3.2.1.1 代码说明
公共模型和接口定义
3.2.1.2 java代码
UserModel.java
public class UserModel {private Long id;private String name;private Integer age;private String birthday;private String address;private String phone;public UserModel() {}public UserModel(Long id, String name, Integer age, String birthday, String address, String phone) {this.id = id;this.name = name;this.age = age;this.birthday = birthday;this.address = address;this.phone = phone;}// get&set&toString
}
UserService.java
public interface UserService {List<UserModel> findAll();UserModel findById(Long id);UserModel add(UserModel userModel);UserModel update(UserModel userModel);UserModel deleteById(Long id);
}
3.2.2 user-service-provider
3.2.1 代码说明
服务提供者，依赖 user-api，实现其中的接口；注册到 eureka server
3.2.2 maven 依赖
<dependencies><dependency><groupId>com.soulballad.usage</groupId><artifactId>user-api</artifactId><version>${project.version}</version></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>
</dependencies>
3.2.3 配置文件
spring.application.name=eureka-client-provider
server.port=8090
eureka.server.host=localhost
eureka.server.port=9090
eureka.client.service-url.defaultZone=http://${eureka.server.host}:${eureka.server.port}/eureka/
3.2.4 java代码
UserRepository.java
@Repository
public class UserRepository {// 预置两条数据，所以起始值从2开始private static final AtomicLong ID_GENERATOR = new AtomicLong(2);// 模拟数据库操作private static final Map<Long, UserModel> USER_MAP = new HashMap<>();@PostConstructpublic void init() {UserModel user1 = new UserModel(1L, "zhangsan", 20, "2000-01-02", "beijing", "13666666666");UserModel user2 = new UserModel(2L, "lisi", 30, "1990-03-23", "shanghai", "13888888888");USER_MAP.put(user1.getId(), user1);USER_MAP.put(user2.getId(), user2);}public List<UserModel> findAll() {return new ArrayList<>(USER_MAP.values());}public UserModel findById(Long id) {return USER_MAP.containsKey(id) ? USER_MAP.get(id) : new UserModel();}public UserModel add(UserModel userModel) {long id = ID_GENERATOR.incrementAndGet();userModel.setId(id);USER_MAP.put(id, userModel);return userModel;}public UserModel update(UserModel userModel) {USER_MAP.put(userModel.getId(), userModel);return USER_MAP.get(userModel.getId());}public UserModel deleteById(Long id) {UserModel userModel = USER_MAP.get(id);USER_MAP.remove(id);return userModel;}
}
UserServiceImpl.java
@Service
public class UserServiceImpl implements UserService {@Autowiredprivate UserRepository userRepository;@Overridepublic List<UserModel> findAll() {return userRepository.findAll();}@Overridepublic UserModel findById(Long id) {return userRepository.findById(id);}@Overridepublic UserModel add(UserModel userModel) {return userRepository.add(userModel);}@Overridepublic UserModel update(UserModel userModel) {return userRepository.update(userModel);}@Overridepublic UserModel deleteById(Long id) {return userRepository.deleteById(id);}
}
UserProviderController.java
@RestController
@RequestMapping(value = "/provider/user")
public class UserProviderController {@Autowiredprivate UserService userService;@GetMapping(value = "/list")public List<UserModel> list() {return userService.findAll();}@GetMapping(value = "/query/{id}")public UserModel query(@PathVariable Long id) {return userService.findById(id);}@PostMapping(value = "/add")public UserModel add(@RequestBody UserModel userModel) {return userService.add(userModel);}@PutMapping(value = "/update")public UserModel update(@RequestBody UserModel userModel) {return userService.update(userModel);}@DeleteMapping(value = "/delete/{id}")public UserModel deleteById(@PathVariable Long id) {return userService.deleteById(id);}
}
3.2.3 user-service-consumer
3.2.3.1 代码说明
服务提供者，依赖 user-api，调用其中的接口；注册到 eureka server
3.2.2 maven 依赖
<dependencies><dependency><groupId>com.soulballad.usage</groupId><artifactId>user-api</artifactId><version>${project.version}</version></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>
</dependencies>
3.2.3 配置文件
spring.application.name=eureka-client-consumer
server.port=8080
eureka.server.host=localhost
eureka.server.port=9090
eureka.client.service-url.defaultZone=http://${eureka.server.host}:${eureka.server.port}/eureka/
3.2.4 java代码
UserServiceProxy.java
@Service
public class UserServiceProxy implements UserService {// user-service-provider 的 application.properties 中 spring.application.name + prefixprivate static final String USER_PROVIDER_PREFIX = "http://eureka-client-provider" + "/provider/user";// 在 UserServiceConsumerApplication 进行声明@Autowiredprivate RestTemplate restTemplate;@Overridepublic List<UserModel> findAll() {UserModel[] userArray = restTemplate.getForObject(USER_PROVIDER_PREFIX + "/list", UserModel[].class);return Arrays.asList(userArray != null ? userArray : new UserModel[0]);}@Overridepublic UserModel findById(Long id) {return restTemplate.getForObject(USER_PROVIDER_PREFIX + "/query/{id}", UserModel.class, id);}@Overridepublic UserModel add(UserModel userModel) {return restTemplate.postForObject(USER_PROVIDER_PREFIX + "/add", userModel, UserModel.class);}@Overridepublic UserModel update(UserModel userModel) {restTemplate.put(USER_PROVIDER_PREFIX + "/update", userModel);return findById(userModel.getId());}@Overridepublic UserModel deleteById(Long id) {UserModel userModel = findById(id);restTemplate.delete(USER_PROVIDER_PREFIX + "/delete/{id}", id);return userModel;}
}
UserConsumerController.java
@RestController
@RequestMapping(value = "/consumer/user")
public class UserConsumerController {@Autowiredprivate UserService userService;@GetMapping(value = "/list")public List<UserModel> list() {return userService.findAll();}@GetMapping(value = "/query/{id}")public UserModel query(@PathVariable Long id) {return userService.findById(id);}@PostMapping(value = "/add")public UserModel add(@RequestBody UserModel userModel) {return userService.add(userModel);}@PutMapping(value = "/update")public UserModel update(@RequestBody UserModel userModel) {return userService.update(userModel);}@DeleteMapping(value = "/delete/{id}")public UserModel deleteById(@PathVariable Long id) {return userService.deleteById(id);}
}
UserServiceConsumerApplication.java
@EnableDiscoveryClient
@SpringBootApplication
public class UserServiceConsumerApplication {public static void main(String[] args) {SpringApplication.run(UserServiceConsumerApplication.class, args);}// 负载均衡@Bean@LoadBalancedpublic RestTemplate restTemplate() {return new RestTemplate();}
}
3.3 git 地址
spring-cloud-nfx-02-eureka: Spring Cloud 整合 Eureka 实现的分布式注册中心方案
4.效果展示
4.1 nfx-eureka-server
启动 eureka 服务端 nfx-eureka-server，访问 http://localhost:9090， 可以看到如下页面
没有任何服务注册到 nfx-eureka-server 上面来。
4.2 user-service-provider
然后再启动 user-service-provider，再次访问 http://localhost:9090，可以看到服务提供者已经注册上来了
在 netflix-eureka-client-provider 访问下列地址，观察输出信息是否符合预期。
查询用户列表
### GET /provider/user/list
GET http://localhost:8090/provider/user/list
Accept: application/json
根据id查询用户
### GET /provider/user/query/{id}
GET http://localhost:8090/provider/user/query/1
Accept: application/json
新增用户
### POST /provider/user/add
POST http://localhost:8090/provider/user/add
Accept: application/json
Content-Type: application/json{"name": "wangwu","age": 20,"birthday": "2000-01-01","address": "wuhan","phone": "15999999999"
}
更新用户
### PUT /provider/user/update
PUT http://localhost:8090/provider/user/update
Accept: application/json
Content-Type: application/json{"id": 2,"name": "lisi","age": 40,"birthday": "1980-01-01","address": "guangzhou","phone": "13888888888"
}
根据id删除用户
### DELETE /provider/user/delete/{id}
DELETE http://localhost:8090/provider/user/delete/3
Accept: application/json
可以看到 user-service-provider 提供的接口都可以正常运行
4.3 user-service-consumer
然后再启动 user-service-consumer，再次访问 http://localhost:9090，可以看到服务提供者已经注册上来了
在 netflix-eureka-client-consumer 访问下列地址，观察输出信息是否符合预期。
查询用户列表
### GET /consumer/user/list
GET http://localhost:8080/consumer/user/list
Accept: application/json
根据id查询用户
### GET /consumer/user/query/{id}
GET http://localhost:8080/consumer/user/query/1
Accept: application/json
新增用户
### POST /consumer/user/add
POST http://localhost:8080/consumer/user/add
Accept: application/json
Content-Type: application/json{"name": "wangwu","age": 20,"birthday": "2000-01-01","address": "wuhan","phone": "15999999999"
}
更新用户
### PUT /consumer/user/update
PUT http://localhost:8080/consumer/user/update
Accept: application/json
Content-Type: application/json{"id": 2,"name": "lisi","age": 40,"birthday": "1980-01-01","address": "shanghang-pudong","phone": "13888888888"
}
根据id删除用户
### DELETE /consumer/user/delete/{id}
DELETE http://localhost:8080/consumer/user/delete/4
Accept: application/json
5.源码分析
5.1 EurekaServer 如何启动？
在使用 @EnableEurekaServer 时，激活了 EurekaServerMarkerConfiguration 配置类，在 EurekaServer 的自动装配类 EurekaServerAutoConfiguration 中，通过构造函数声明了 EurekaController
@Bean
@ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled",matchIfMissing = true)
public EurekaController eurekaController() {return new EurekaController(this.applicationInfoManager);
}
这个 EurekaController 实际上就是访问 http://localhost:9090/ 时对应的 eureka 管理后台，它使用spring-mvc 来进行实现。
@Controller
@RequestMapping("${eureka.dashboard.path:/}")
public class EurekaController {@Value("${eureka.dashboard.path:/}")private String dashboardPath = "";private ApplicationInfoManager applicationInfoManager;public EurekaController(ApplicationInfoManager applicationInfoManager) {this.applicationInfoManager = applicationInfoManager;}@RequestMapping(method = RequestMethod.GET)public String status(HttpServletRequest request, Map<String, Object> model) {populateBase(request, model);populateApps(model);StatusInfo statusInfo;try {statusInfo = new StatusResource().getStatusInfo();}catch (Exception e) {statusInfo = StatusInfo.Builder.newBuilder().isHealthy(false).build();}model.put("statusInfo", statusInfo);populateInstanceInfo(model, statusInfo);filterReplicas(model, statusInfo);return "eureka/status";}
}
可以看到 spring-cloud-starter-netflix-eureka-server 的依赖关系如下
5.2 EurekaClient如何注册？
服务启动的时候，会在刷新上下文的时候启动 Lifecycle，EurekaAutoServiceRegistration 是 Lifecycle 的一个实现类，所以会调用它的 start 方法，在 start 方法中通过调用 serviceRegistry.register 方法来进行注册。
这里的 serviceRegistry 是 EurekaServiceRegistry，EurekaServiceRegistry 实现了 ServiceRegistry 接口，ServiceRegistry 在 spring-cloud-common 中进行定义，它是一个通用的接口，根据实现方案的不同，它还可以是 ConsulServiceRegistry、NacosServiceRegistry、ZookeeperServiceRegistry 等。
EurekaServiceRegistry#register
@Override
public void register(EurekaRegistration reg) {maybeInitializeClient(reg);if (log.isInfoEnabled()) {log.info("Registering application "+ reg.getApplicationInfoManager().getInfo().getAppName()+ " with eureka with status "+ reg.getInstanceConfig().getInitialStatus());}// 设置实例状态reg.getApplicationInfoManager().setInstanceStatus(reg.getInstanceConfig().getInitialStatus());// 健康检查reg.getHealthCheckHandler().ifAvailable(healthCheckHandler -> reg.getEurekaClient().registerHealthCheck(healthCheckHandler));
}
在 setInstanceStatus 中调用 listener.notify 进行通知
public synchronized void setInstanceStatus(InstanceStatus status) {InstanceStatus next = instanceStatusMapper.map(status);if (next == null) {return;}InstanceStatus prev = instanceInfo.setStatus(next);if (prev != null) {for (StatusChangeListener listener : listeners.values()) {try {listener.notify(new StatusChangeEvent(prev, next));} catch (Exception e) {logger.warn("failed to notify listener: {}", listener.getId(), e);}}}
}
这里的 listeners 通过 registerStatusChangeListener 方法进行注册
public void registerStatusChangeListener(StatusChangeListener listener) {listeners.put(listener.getId(), listener);
}
这个方法在 DiscoveryClient 中进行调用
private ApplicationInfoManager.StatusChangeListener statusChangeListener;
statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {@Overridepublic String getId() {return "statusChangeListener";}@Overridepublic void notify(StatusChangeEvent statusChangeEvent) {if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {// log at warn level if DOWN was involvedlogger.warn("Saw local status change event {}", statusChangeEvent);} else {logger.info("Saw local status change event {}", statusChangeEvent);}instanceInfoReplicator.onDemandUpdate();}
};if (clientConfig.shouldOnDemandUpdateStatusChange()) {applicationInfoManager.registerStatusChangeListener(statusChangeListener);
}
这里的 listener 为 ApplicationInfoManager.StatusChangeListener，所以调用到它的 notify 方法；然后调用到
instanceInfoReplicator.onDemandUpdate()
public boolean onDemandUpdate() {if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {if (!scheduler.isShutdown()) {// 开启一个任务scheduler.submit(new Runnable() {@Overridepublic void run() {logger.debug("Executing on-demand update of local InstanceInfo");Future latestPeriodic = scheduledPeriodicRef.get();if (latestPeriodic != null && !latestPeriodic.isDone()) {logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");latestPeriodic.cancel(false);}InstanceInfoReplicator.this.run();}});return true;} else {logger.warn("Ignoring onDemand update due to stopped scheduler");return false;}} else {logger.warn("Ignoring onDemand update due to rate limiter");return false;}
}
最终调用到 InstanceInfoReplicator.this.run()，通过 discoveryClient.register 进行注册
public void run() {try {// 刷新实例信息discoveryClient.refreshInstanceInfo();Long dirtyTimestamp = instanceInfo.isDirtyWithTime();if (dirtyTimestamp != null) {// 注册discoveryClient.register();instanceInfo.unsetIsDirty(dirtyTimestamp);}} catch (Throwable t) {logger.warn("There was a problem with the instance info replicator", t);} finally {Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);scheduledPeriodicRef.set(next);}
}
boolean register() throws Throwable {logger.info(PREFIX + "{}: registering service...", appPathIdentifier);EurekaHttpResponse<Void> httpResponse;try {httpResponse = eurekaTransport.registrationClient.register(instanceInfo);} catch (Exception e) {logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);throw e;}if (logger.isInfoEnabled()) {logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());}return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
}
这里使用 jersey 进行了 http 调用，发送 post 请求
@Override
public EurekaHttpResponse<Void> register(InstanceInfo info) {// 请求路径为 apps/EUREKA-CLIENT-PROVIDER 或 apps/EUREKA-CLIENT-CONSUMERString urlPath = "apps/" + info.getAppName();ClientResponse response = null;try {Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();addExtraHeaders(resourceBuilder);// Post 请求，媒体类型是 jsonresponse = resourceBuilder.header("Accept-Encoding", "gzip").type(MediaType.APPLICATION_JSON_TYPE).accept(MediaType.APPLICATION_JSON).post(ClientResponse.class, info);return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();} finally {if (logger.isDebugEnabled()) {logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),response == null ? "N/A" : response.getStatus());}if (response != null) {response.close();}}
}
5.3 EurekaServer处理注册请求
client 端发送请求后到达 ApplicationResource#addInstance
@POST
@Consumes({"application/json", "application/xml"})
public Response addInstance(InstanceInfo info,@HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);// validate that the instanceinfo contains all the necessary required fields// 参数校验if (isBlank(info.getId())) {return Response.status(400).entity("Missing instanceId").build();} else if (isBlank(info.getHostName())) {return Response.status(400).entity("Missing hostname").build();} else if (isBlank(info.getIPAddr())) {return Response.status(400).entity("Missing ip address").build();} else if (isBlank(info.getAppName())) {return Response.status(400).entity("Missing appName").build();} else if (!appName.equals(info.getAppName())) {return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();} else if (info.getDataCenterInfo() == null) {return Response.status(400).entity("Missing dataCenterInfo").build();} else if (info.getDataCenterInfo().getName() == null) {return Response.status(400).entity("Missing dataCenterInfo Name").build();}// handle cases where clients may be registering with bad DataCenterInfo with missing dataDataCenterInfo dataCenterInfo = info.getDataCenterInfo();if (dataCenterInfo instanceof UniqueIdentifier) {String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();if (isBlank(dataCenterInfoId)) {boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));if (experimental) {String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";return Response.status(400).entity(entity).build();} else if (dataCenterInfo instanceof AmazonInfo) {AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);if (effectiveId == null) {amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());}} else {logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());}}}// 注册registry.register(info, "true".equals(isReplication));return Response.status(204).build(); // 204 to be backwards compatible
}
这里的 registry 是 PeerAwareInstanceRegistry，它的类图如下
最终调用的 register 方法在 AbstractInstanceRegistry 中
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {try {read.lock();// 从registry中获取当前app的实例信息mapMap<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());// 增加注册次数REGISTER.increment(isReplication);// 如果是第一次注册，初始化一个ConcurrentHashMapif (gMap == null) {final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);if (gMap == null) {gMap = gNewMap;}}// 从gMap中获取存在的Lease信息Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());// Retain the last dirty timestamp without overwriting it, if there is already a leaseif (existingLease != null && (existingLease.getHolder() != null)) {Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);// this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted// InstanceInfo instead of the server local copy.if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +" than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");registrant = existingLease.getHolder();}} else {// The lease does not exist and hence it is a new registrationsynchronized (lock) {if (this.expectedNumberOfClientsSendingRenews > 0) {// Since the client wants to register it, increase the number of clients sending renewsthis.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;updateRenewsPerMinThreshold();}}logger.debug("No previous lease information found; it is new registration");}// 构建一个LeaseLease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);if (existingLease != null) {lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());}gMap.put(registrant.getId(), lease);synchronized (recentRegisteredQueue) {recentRegisteredQueue.add(new Pair<Long, String>(System.currentTimeMillis(),registrant.getAppName() + "(" + registrant.getId() + ")"));}// This is where the initial state transfer of overridden status happensif (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "+ "overrides", registrant.getOverriddenStatus(), registrant.getId());if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {logger.info("Not found overridden id {} and hence adding it", registrant.getId());overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());}}InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());if (overriddenStatusFromMap != null) {logger.info("Storing overridden status {} from map", overriddenStatusFromMap);registrant.setOverriddenStatus(overriddenStatusFromMap);}// Set the status based on the overridden status rulesInstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);registrant.setStatusWithoutDirty(overriddenInstanceStatus);// 得到Lease实例，判断状态是否为UPif (InstanceStatus.UP.equals(registrant.getStatus())) {lease.serviceUp();}// 设置注册类型为新增registrant.setActionType(ActionType.ADDED);recentlyChangedQueue.add(new RecentlyChangedItem(lease));registrant.setLastUpdatedTimestamp();// 缓存过期invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());logger.info("Registered instance {}/{} with status {} (replication={})",registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);} finally {read.unlock();}
}
在 register 完成之后，有一个 replicateToPeers() 方法，它用来实现集群节点之间信息复制
private void replicateToPeers(Action action, String appName, String id,InstanceInfo info /* optional */,InstanceStatus newStatus /* optional */, boolean isReplication) {Stopwatch tracer = action.getTimer().start();try {if (isReplication) {numberOfReplicationsLastMin.increment();}// If it is a replication already, do not replicate again as this will create a poison replicationif (peerEurekaNodes == Collections.EMPTY_LIST || isReplication) {return;}// 获取到所有的nodefor (final PeerEurekaNode node : peerEurekaNodes.getPeerEurekaNodes()) {// If the url represents this host, do not replicate to yourself.if (peerEurekaNodes.isThisMyUrl(node.getServiceUrl())) {continue;}// 复制实例信息到每个nodereplicateInstanceActionsToPeers(action, appName, id, info, newStatus, node);}} finally {tracer.stop();}
}
replicateInstanceActionsToPeers 实现如下
private void replicateInstanceActionsToPeers(Action action, String appName,String id, InstanceInfo info, InstanceStatus newStatus,PeerEurekaNode node) {try {InstanceInfo infoFromRegistry = null;CurrentRequestVersion.set(Version.V2);// 判断操作类型switch (action) { case Cancel:// 取消注册node.cancel(appName, id);break;case Heartbeat:// 心跳InstanceStatus overriddenStatus = overriddenInstanceStatusMap.get(id);infoFromRegistry = getInstanceByAppAndId(appName, id, false);node.heartbeat(appName, id, infoFromRegistry, overriddenStatus, false);break;case Register:// 注册node.register(info);break;case StatusUpdate:// 状态变更infoFromRegistry = getInstanceByAppAndId(appName, id, false);node.statusUpdate(appName, id, newStatus, infoFromRegistry);break;case DeleteStatusOverride:// 删除被重写的实例infoFromRegistry = getInstanceByAppAndId(appName, id, false);node.deleteStatusOverride(appName, id, infoFromRegistry);break;}} catch (Throwable t) {logger.error("Cannot replicate information to {} for action {}", node.getServiceUrl(), action.name(), t);}
}
6.参考
- 官方文档-Spring Cloud Netflix/Ribbon
2.SpringCloud学习(二)——Spring Cloud Eureka 服务注册中心相关推荐
- eureka集群只注册一个_一、Spring Cloud Eureka服务注册中心
Sping boot Eureka是服务中心,管理各种服务功能包括服务的注册.发现.熔断.负载.降级等.Spring Cloud Eureka 是对Netflix公司的Eureka的二次封装. 在du ...
- 如何优化Spring Cloud微服务注册中心架构？
作者: 石杉的架构笔记 1.再回顾:什么是服务注册中心? 先回顾一下什么叫做服务注册中心? 顾名思义,假设你有一个分布式系统,里面包含了多个服务,部署在不同的机器上,然后这些不同机器上的服务之间要互相 ...
- Spring Cloud 学习笔记(一) 之服务治理模块Spring Cloud Eureka 搭建注册中心
2019独角兽企业重金招聘Python工程师标准>>> 关于springboot的学习请参考前面的文章 接下来我们会开启一系列关于springcloud的学习文章. 一.概念 首先我 ...
- spring cloud eureka服务注册和调用
SPRING INITIALIZR构建工程 spring boot 可通过SPRING INITIALIZR构建项目 访问SPRING INITIALIZR官网,填写项目相关信息后,生成项目. 将下载 ...
- spring cloud微服务注册中心EurekaServer
注册中心EurekaServer pom文件: <?xml version="1.0" encoding="UTF-8"?> <project ...
- 新 Spring Cloud (一) 之 Eureka 服务注册中心
文章目录 一.前言 0. 之前写过两篇Spring Cloud,但是感觉不够具体,所以重新写了一份. 1. SpringCloud 2. 什么是Eureka 3. 原理图 二.基本使用实例 1. 场景 ...
- 天荒地老修仙功-第六部：Spring Cloud Eureka——服务发现
文章目录 前言 1.Eureka 1.1.Spring Cloud Netflix 1.2.样例项目结构 1.3.服务注册中心 1.4.客户端(服务提供者) 1.5.高可用 1.5.1.Region ...
- SpringCloud笔记(1)—Eureka服务注册中心
项目开发练习涉及的模块 Eureka:负责服务发现相关.Eureka Server/Client.高可用.服务发现机制. Config:负责统一配置中心.Config Server/Client.Sp ...
- SpringCloud笔记(Hoxton)——Netflix之Eureka服务注册与发现
基础应用 Eureka简介 Eureka是一个基于REST的服务,主要在AWS云中使用,定位服务来进行中间层服务器的负载均衡和故障转移. SpringCloud封装了Netflix公司开发的Eurek ...
- SpringCloud系列（一）、服务注册中心Eureka基础
启动Eureka服务注册中心 1.微服务的注册中心 1.1.注册中心的主要作用 1.2.常见的注册中心 1.3.常见注册中心的异同点 2.Eureka概述 2.1.Eureka的基础知识 2.2.Eu ...
最新文章
- kafka channle的应用案例
- vue如何使用原生js写动画效果_深入理解 Vuejs 动画效果
- new 实例化对象是啥意思_二. 初步认识JS中的类和对象
- MySQL命令（二）| 表的增删查改、聚合函数（复合函数）、联合查询
- 免费开源的thinkphp办公管理系统
- React Native应用实现步骤
- ipython介绍及使用
- oracle数据库三大日志,Oracle 数据库日志和用户日志位置
- roseha 8.9 for linux oracle 11g,RoseHA 9.0 for Linux配合Oracle12c配置文档_v2.0-2015-04.pdf
- 深度学习常用框架和基础模型
- java long型时间戳_深入理解java long 存储时间戳
- 简单翻译工具--必应词典第三方api使用方法
- RL(Chapter 6): Cliff Walking
- 熊孩子说“你没看过奥特曼”，赶紧用Python学习一下，没想到
- 运维的职业发展方向有哪些？该如何规划？
- 计算机专业论文谢辞,计算机专业论文致谢信.docx
- 050002《马云说》读后感——勇者无畏
- 莫比乌斯反演小结 + 黑暗爆炸 2301
- Java服务器页面：JSP
- python常量基本类型有哪些_【Python③】python基本数据类型，变量和常量
热门文章
- win7升级win10正式版_如何在win7系统中升级win10
- 计算机网络软件系统不包括,1period;计算机软件系统一般分为lpar; A rpar;两大部分...
- 怎么把电脑上的python软件卸载干净_怎么把一个软件卸载干净 把一个软件卸载干净的两种方法...
- 管理储存组和邮箱数据库
- 组织分解结构(Organizational Breakdown Structure OBS)
- FastAdmin 文件图片上传失败File size error
- 做一个功能模块的实现流程
- instant-ngp总结
- 豆瓣高分电影爬取项目，思路清晰，易上手，附免费代码下载
- 【TL学习笔记】1：领域自适应(Domain Adaptation)方法综述