msb-public
/
hippo4j
mirror of https://github.com/longtai-cn/hippo4j
1
0
Code Issues Projects Releases Wiki Activity

服务注册核心逻辑重构. (#91)

Browse Source
pull/106/head
chen.ma 3 years ago
parent 7936dd5ea4
commit 7cc3b42ad2
5 changed files with 61 additions and 246 deletions
Show Stats Download Patch File Download Diff File

6
hippo4j-discovery/src/main/java/cn/hippo4j/discovery/controller/ApplicationController.java
Unescape View File

@ -6,8 +6,7 @@ import cn.hippo4j.common.web.base.Results;
import cn.hippo4j.common.web.exception.ErrorCodeEnum;
import cn.hippo4j.discovery.core.InstanceRegistry;
import cn.hippo4j.discovery.core.Lease;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.*;
@ -23,11 +22,10 @@ import static cn.hippo4j.common.constant.Constants.BASE_PATH;
*/
@Slf4j
@RestController
@RequiredArgsConstructor
@AllArgsConstructor
@RequestMapping(BASE_PATH + "/apps")
public class ApplicationController {
@NonNull
private final InstanceRegistry instanceRegistry;
@GetMapping("/{appName}")

227
hippo4j-discovery/src/main/java/cn/hippo4j/discovery/core/BaseInstanceRegistry.java
Unescape View File

@ -3,19 +3,22 @@ package cn.hippo4j.discovery.core;
import cn.hippo4j.common.design.observer.AbstractSubjectCenter;
import cn.hippo4j.common.model.InstanceInfo;
import cn.hippo4j.common.model.InstanceInfo.InstanceStatus;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import java.util.*;
import java.util.concurrent.*;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static cn.hippo4j.common.constant.Constants.EVICTION_INTERVAL_TIMER_IN_MS;
import static cn.hippo4j.common.constant.Constants.SCHEDULED_THREAD_CORE_NUM;
@ -23,6 +26,11 @@ import static cn.hippo4j.common.constant.Constants.SCHEDULED_THREAD_CORE_NUM;
/**
* Base instance registry.
*
* <p>
* Reference from Eureka.
* Service registration, service offline, service renewal.
* </p>
*
* @author chen.ma
* @date 2021/8/8 22:46
*/
@ -30,33 +38,9 @@ import static cn.hippo4j.common.constant.Constants.SCHEDULED_THREAD_CORE_NUM;
@Service
public class BaseInstanceRegistry implements InstanceRegistry<InstanceInfo> {
private final int CONTAINER_SIZE = 1024;
private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private final Lock read = readWriteLock.readLock();
protected final Object lock = new Object();
private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap(CONTAINER_SIZE);
protected volatile int expectedNumberOfClientsSendingRenews;
private final CircularQueue<Pair<Long, String>> recentRegisteredQueue;
private final CircularQueue<Pair<Long, String>> recentCanceledQueue;
private final int containerSize = 1024;
private ConcurrentLinkedQueue<RecentlyChangedItem> recentlyChangedQueue = new ConcurrentLinkedQueue();
protected final ConcurrentMap<String, InstanceStatus> overriddenInstanceStatusMap = CacheBuilder
.newBuilder().initialCapacity(512)
.expireAfterAccess(1, TimeUnit.HOURS)
.<String, InstanceStatus>build().asMap();
public BaseInstanceRegistry() {
this.recentRegisteredQueue = new CircularQueue(CONTAINER_SIZE);
this.recentCanceledQueue = new CircularQueue(CONTAINER_SIZE);
}
private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap(containerSize);
@Override
public List<Lease<InstanceInfo>> listInstance(String appName) {
@ -71,54 +55,38 @@ public class BaseInstanceRegistry implements InstanceRegistry<InstanceInfo> {
}
@Override
public void register(InstanceInfo registrant) {
read.lock();
try {
Map<String, Lease<InstanceInfo>> registerMap = registry.get(registrant.getAppName());
public synchronized void register(InstanceInfo registrant) {
Map<String, Lease<InstanceInfo>> registerMap = registry.get(registrant.getAppName());
if (registerMap == null) {
ConcurrentHashMap<String, Lease<InstanceInfo>> registerNewMap = new ConcurrentHashMap(12);
registerMap = registry.putIfAbsent(registrant.getAppName(), registerNewMap);
if (registerMap == null) {
ConcurrentHashMap<String, Lease<InstanceInfo>> registerNewMap = new ConcurrentHashMap(12);
registerMap = registry.putIfAbsent(registrant.getAppName(), registerNewMap);
if (registerMap == null) {
registerMap = registerNewMap;
}
}
Lease<InstanceInfo> existingLease = registerMap.get(registrant.getInstanceId());
if (existingLease != null && (existingLease.getHolder() != null)) {
Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
registrant = existingLease.getHolder();
}
registerMap = registerNewMap;
}
}
Lease<InstanceInfo> lease = new Lease(registrant);
if (existingLease != null) {
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
registerMap.put(registrant.getInstanceId(), lease);
recentRegisteredQueue.add(new Pair(
System.currentTimeMillis(),
registrant.getAppName() + "(" + registrant.getInstanceId() + ")"));
Lease<InstanceInfo> existingLease = registerMap.get(registrant.getInstanceId());
if (existingLease != null && (existingLease.getHolder() != null)) {
Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getInstanceId());
if (overriddenStatusFromMap != null) {
log.info("Storing overridden status :: {} from map", overriddenStatusFromMap);
registrant.setOverriddenStatus(overriddenStatusFromMap);
if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
registrant = existingLease.getHolder();
}
}
if (InstanceStatus.UP.equals(registrant.getStatus())) {
lease.serviceUp();
}
Lease<InstanceInfo> lease = new Lease(registrant);
if (existingLease != null) {
lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
}
registerMap.put(registrant.getInstanceId(), lease);
registrant.setActionType(InstanceInfo.ActionType.ADDED);
recentlyChangedQueue.add(new RecentlyChangedItem(lease));
registrant.setLastUpdatedTimestamp();
} finally {
read.unlock();
if (InstanceStatus.UP.equals(registrant.getStatus())) {
lease.serviceUp();
}
registrant.setActionType(InstanceInfo.ActionType.ADDED);
registrant.setLastUpdatedTimestamp();
}
@Override
@ -127,7 +95,7 @@ public class BaseInstanceRegistry implements InstanceRegistry<InstanceInfo> {
String instanceId = instanceRenew.getInstanceId();
Map<String, Lease<InstanceInfo>> registryMap = registry.get(appName);
Lease<InstanceInfo> leaseToRenew = null;
Lease<InstanceInfo> leaseToRenew;
if (registryMap == null || (leaseToRenew = registryMap.get(instanceId)) == null) {
return false;
}
@ -138,94 +106,21 @@ public class BaseInstanceRegistry implements InstanceRegistry<InstanceInfo> {
@Override
public void remove(InstanceInfo info) {
ReentrantReadWriteLock.WriteLock writeLock = readWriteLock.writeLock();
writeLock.lock();
try {
String appName = info.getAppName();
String instanceId = info.getInstanceId();
Map<String, Lease<InstanceInfo>> leaseMap = registry.get(appName);
if (CollectionUtils.isEmpty(leaseMap)) {
log.warn("Failed to remove unhealthy node, no application found :: {}", appName);
return;
}
Lease<InstanceInfo> remove = leaseMap.remove(instanceId);
if (remove == null) {
log.warn("Failed to remove unhealthy node, no instance found :: {}", instanceId);
return;
}
log.info("Remove unhealthy node, node ID :: {}", instanceId);
} finally {
writeLock.unlock();
String appName = info.getAppName();
String instanceId = info.getInstanceId();
Map<String, Lease<InstanceInfo>> leaseMap = registry.get(appName);
if (CollectionUtils.isEmpty(leaseMap)) {
log.warn("Failed to remove unhealthy node, no application found :: {}", appName);
return;
}
}
static class CircularQueue<E> extends AbstractQueue<E> {
private final ArrayBlockingQueue<E> delegate;
public CircularQueue(int capacity) {
this.delegate = new ArrayBlockingQueue(capacity);
}
@Override
public Iterator<E> iterator() {
return delegate.iterator();
}
@Override
public int size() {
return delegate.size();
}
@Override
public boolean offer(E e) {
while (!delegate.offer(e)) {
delegate.poll();
}
return true;
}
@Override
public E poll() {
return delegate.poll();
}
@Override
public E peek() {
return delegate.peek();
}
@Override
public void clear() {
delegate.clear();
}
@Override
public Object[] toArray() {
return delegate.toArray();
Lease<InstanceInfo> remove = leaseMap.remove(instanceId);
if (remove == null) {
log.warn("Failed to remove unhealthy node, no instance found :: {}", instanceId);
return;
}
}
private static final class RecentlyChangedItem {
private long lastUpdateTime;
private Lease<InstanceInfo> leaseInfo;
public RecentlyChangedItem(Lease<InstanceInfo> lease) {
this.leaseInfo = lease;
lastUpdateTime = System.currentTimeMillis();
}
public long getLastUpdateTime() {
return this.lastUpdateTime;
}
public Lease<InstanceInfo> getLeaseInfo() {
return this.leaseInfo;
}
log.info("Remove unhealthy node, node ID :: {}", instanceId);
}
public void evict(long additionalLeaseMs) {
@ -246,23 +141,17 @@ public class BaseInstanceRegistry implements InstanceRegistry<InstanceInfo> {
String appName = expiredLease.getHolder().getAppName();
String id = expiredLease.getHolder().getInstanceId();
String identify = expiredLease.getHolder().getIdentify();
internalCancel(appName, id, identify, false);
internalCancel(appName, id, identify);
}
}
protected boolean internalCancel(String appName, String id, String identify, boolean isReplication) {
read.lock();
try {
Map<String, Lease<InstanceInfo>> registerMap = registry.get(appName);
if (!CollectionUtils.isEmpty(registerMap)) {
registerMap.remove(id);
AbstractSubjectCenter.notify(AbstractSubjectCenter.SubjectType.CLEAR_CONFIG_CACHE, () -> identify);
log.info("Clean up unhealthy nodes. Node id :: {}", id);
}
protected boolean internalCancel(String appName, String id, String identify) {
Map<String, Lease<InstanceInfo>> registerMap = registry.get(appName);
if (!CollectionUtils.isEmpty(registerMap)) {
registerMap.remove(id);
AbstractSubjectCenter.notify(AbstractSubjectCenter.SubjectType.CLEAR_CONFIG_CACHE, () -> identify);
} finally {
read.unlock();
log.info("Clean up unhealthy nodes. Node id :: {}", id);
}
return true;

36
hippo4j-discovery/src/main/java/cn/hippo4j/discovery/core/Pair.java
Unescape View File

@ -1,36 +0,0 @@
package cn.hippo4j.discovery.core;
/**
* Pair.
*
* @author chen.ma
* @date 2021/8/8 23:04
*/
public class Pair<E1, E2> {
public E1 first() {
return first;
}
public void setFirst(E1 first) {
this.first = first;
}
public E2 second() {
return second;
}
public void setSecond(E2 second) {
this.second = second;
}
private E1 first;
private E2 second;
public Pair(E1 first, E2 second) {
this.first = first;
this.second = second;
}
}

36
hippo4j-discovery/src/main/java/cn/hippo4j/discovery/core/StatusOverrideResult.java
Unescape View File

@ -1,36 +0,0 @@
package cn.hippo4j.discovery.core;
import cn.hippo4j.common.model.InstanceInfo.InstanceStatus;
/**
* Status override result.
*
* @author chen.ma
* @date 2021/8/8 23:11
*/
public class StatusOverrideResult {
public static StatusOverrideResult NO_MATCH = new StatusOverrideResult(false, null);
public static StatusOverrideResult matchingStatus(InstanceStatus status) {
return new StatusOverrideResult(true, status);
}
private final boolean matches;
private final InstanceStatus status;
private StatusOverrideResult(boolean matches, InstanceStatus status) {
this.matches = matches;
this.status = status;
}
public boolean matches() {
return matches;
}
public InstanceStatus status() {
return status;
}
}

2
hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/starter/config/DiscoveryConfiguration.java
Unescape View File

@ -40,7 +40,7 @@ public class DiscoveryConfiguration {
String namespace = properties.getNamespace();
String itemId = properties.getItemId();
String port = environment.getProperty("server.port");
String applicationName = environment.getProperty("spring.application.name");
String applicationName = environment.getProperty("spring.dynamic.thread-pool.item-id");
String active = environment.getProperty("spring.profiles.active", "UNKNOWN");
InstanceInfo instanceInfo = new InstanceInfo();

Write Preview
Loading…
Cancel
Save