spring cloud手动下掉节点
systemctl restart xxxxRuntime.getRuntime().addShutdownHook(new Thread(() -> {
try {
doSomeThing();
} catch (Exception ex) {
log.error("doSomeThing error. ", ex);
}
}));最后更新于
class SpringApplicationShutdownHook implements Runnable {
// ...
void registerApplicationContext(ConfigurableApplicationContext context) {
addRuntimeShutdownHookIfNecessary();
synchronized (SpringApplicationShutdownHook.class) {
assertNotInProgress();
context.addApplicationListener(this.contextCloseListener);
this.contexts.add(context);
}
}
private void addRuntimeShutdownHookIfNecessary() {
if (this.shutdownHookAdded.compareAndSet(false, true)) {
addRuntimeShutdownHook();
}
}
void addRuntimeShutdownHook() {
try {
Runtime.getRuntime().addShutdownHook(new Thread(this, "SpringApplicationShutdownHook"));
}
catch (AccessControlException ex) {
// Not allowed in some environments
}
}
@Override
public void run() {
Set<ConfigurableApplicationContext> contexts;
Set<ConfigurableApplicationContext> closedContexts;
Set<Runnable> actions;
synchronized (SpringApplicationShutdownHook.class) {
this.inProgress = true;
contexts = new LinkedHashSet<>(this.contexts);
closedContexts = new LinkedHashSet<>(this.closedContexts);
actions = new LinkedHashSet<>(this.handlers.getActions());
}
contexts.forEach(this::closeAndWait);
closedContexts.forEach(this::closeAndWait);
actions.forEach(Runnable::run);
}
//...
}public abstract class AbstractAutoServiceRegistration<R extends Registration>
implements AutoServiceRegistration, ApplicationContextAware, ApplicationListener<WebServerInitializedEvent> {
// ...
public void onApplicationEvent(WebServerInitializedEvent event) {
bind(event);
}
@Deprecated
public void bind(WebServerInitializedEvent event) {
ApplicationContext context = event.getApplicationContext();
if (context instanceof ConfigurableWebServerApplicationContext) {
if ("management".equals(((ConfigurableWebServerApplicationContext) context).getServerNamespace())) {
return;
}
}
this.port.compareAndSet(0, event.getWebServer().getPort());
this.start();
}
// ...
public void start() {
if (!isEnabled()) {
if (logger.isDebugEnabled()) {
logger.debug("Discovery Lifecycle disabled. Not starting");
}
return;
}
// only initialize if nonSecurePort is greater than 0 and it isn't already running
// because of containerPortInitializer below
if (!this.running.get()) {
this.context.publishEvent(new InstancePreRegisteredEvent(this, getRegistration()));
register();
if (shouldRegisterManagement()) {
registerManagement();
}
this.context.publishEvent(new InstanceRegisteredEvent<>(this, getConfiguration()));
this.running.compareAndSet(false, true);
}
}
// ...
protected void register() {
this.serviceRegistry.register(getRegistration());
}
// ...
@PreDestroy
public void destroy() {
stop();
}
// ...
public void stop() {
if (this.getRunning().compareAndSet(true, false) && isEnabled()) {
deregister();
if (shouldRegisterManagement()) {
deregisterManagement();
}
this.serviceRegistry.close();
}
}
server:
port: 8801
shutdown: graceful
spring:
lifecycle:
timeout-per-shutdown-phase: 30spublic class NacosDiscoveryClient implements DiscoveryClient {
// ...
@Override
public List<ServiceInstance> getInstances(String serviceId) {
try {
return Optional.of(serviceDiscovery.getInstances(serviceId)).map(instances -> {
ServiceCache.setInstances(serviceId, instances);
return instances;
}).get();
}
catch (Exception e) {
if (failureToleranceEnabled) {
return ServiceCache.getInstances(serviceId);
}
throw new RuntimeException(
"Can not get hosts from nacos server. serviceId: " + serviceId, e);
}
}
// ...
}public class NacosServiceDiscovery {
// ...
public List<ServiceInstance> getInstances(String serviceId) throws NacosException {
String group = discoveryProperties.getGroup();
List<Instance> instances = namingService().selectInstances(serviceId, group,
true);
return hostToServiceInstanceList(instances, serviceId);
}
// ...
private NamingService namingService() {
return nacosServiceManager
.getNamingService(discoveryProperties.getNacosProperties());
}
}@SuppressWarnings("PMD.ServiceOrDaoClassShouldEndWithImplRule")
public class NacosNamingService implements NamingService {
// ...
@Override
public List<Instance> selectInstances(String serviceName, String groupName, boolean healthy) throws NacosException {
return selectInstances(serviceName, groupName, healthy, true);
}
// ...
@Override
public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy,
boolean subscribe) throws NacosException {
ServiceInfo serviceInfo;
if (subscribe) { // 走到这个分支
serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
StringUtils.join(clusters, ","));
} else {
serviceInfo = hostReactor
.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
StringUtils.join(clusters, ","));
}
return selectInstances(serviceInfo, healthy);
}
// ...
}
public class HostReactor implements Closeable {
// ...
public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
String key = ServiceInfo.getKey(serviceName, clusters);
if (failoverReactor.isFailoverSwitch()) {
return failoverReactor.getService(key);
}
ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
if (null == serviceObj) {
serviceObj = new ServiceInfo(serviceName, clusters);
serviceInfoMap.put(serviceObj.getKey(), serviceObj);
updatingMap.put(serviceName, new Object());
updateServiceNow(serviceName, clusters);
updatingMap.remove(serviceName);
} else if (updatingMap.containsKey(serviceName)) {
if (UPDATE_HOLD_INTERVAL > 0) {
// hold a moment waiting for update finish
synchronized (serviceObj) {
try {
serviceObj.wait(UPDATE_HOLD_INTERVAL);
} catch (InterruptedException e) {
NAMING_LOGGER
.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
}
}
}
}
scheduleUpdateIfAbsent(serviceName, clusters); // 定时更新
return serviceInfoMap.get(serviceObj.getKey());
}
// ...
private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
String key = ServiceInfo.getKey(serviceName, clusters);
return serviceInfoMap.get(key);
}
// ...
public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
}
synchronized (futureMap) {
if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
return;
}
ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
}
}
// ...
public synchronized ScheduledFuture<?> addTask(UpdateTask task) {
return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
}
public HostReactor(NamingProxy serverProxy, BeatReactor beatReactor, String cacheDir) {
this(serverProxy, beatReactor, cacheDir, false, false, UtilAndComs.DEFAULT_POLLING_THREAD_COUNT);
}
public HostReactor(NamingProxy serverProxy, BeatReactor beatReactor, String cacheDir, boolean loadCacheAtStart,
boolean pushEmptyProtection, int pollingThreadCount) {
// init executorService
this.executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setDaemon(true);
thread.setName("com.alibaba.nacos.client.naming.updater");
return thread;
}
});
// ...
}
}import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.client.serviceregistry.Registration;
import org.springframework.cloud.client.serviceregistry.ServiceRegistry;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class DeregisterController {
@Autowired
private Registration registration;
@Autowired
private ServiceRegistry serviceRegistry;
@RequestMapping("/deregister")
public String deregister() {
serviceRegistry.deregister(registration);
return "ok";
}
}