【架構師系列】QConfig配置中心系列之Server端（三）
聲明
原創文章，轉載請標注。http://www.rzrgm.cn/boycelee/p/18055933
《碼頭工人的一千零一夜》是一位專注于技術干貨分享的博主，追隨博主的文章，你將深入了解業界最新的技術趨勢，以及在Java開發和安全領域的實用經驗分享。無論你是開發人員還是對逆向工程感興趣的愛好者，都能在《碼頭工人的一千零一夜》找到有價值的知識和見解。
配置中心系列文章
《【架構師視角系列】風控場景下的配置中心設計思考》 http://www.rzrgm.cn/boycelee/p/18355942
《【架構師視角系列】Apollo配置中心之架構設計（一）》http://www.rzrgm.cn/boycelee/p/17967590
《【架構師視角系列】Apollo配置中心之Client端（二）》http://www.rzrgm.cn/boycelee/p/17978027
《【架構師視角系列】Apollo配置中心之Server端（ConfigSevice）（三）》http://www.rzrgm.cn/boycelee/p/18005318
《【架構師視角系列】QConfig配置中心系列之架構設計（一）》http://www.rzrgm.cn/boycelee/p/18013653
《【架構師視角系列】QConfig配置中心系列之Client端（二）》http://www.rzrgm.cn/boycelee/p/18033286
《【架構師視角系列】QConfig配置中心系列之Server端（三）》http://www.rzrgm.cn/boycelee/p/18055933
一、通知與配置拉取
二、設計思考
1、Admin如何通知Server所有實例配置發生變更？
2、Server如何通知Client端配置發生變更？
3、Client如何拉取配置？
三、源碼分析
1、Admin配置推送
1.1、主動推送
1.1.1、邏輯描述
QConfig的Server配置發現有兩種方式，一種是主動推送，另一種是被動掃描。
主動發現是Admin（管理平臺）通過注冊中心獲取到已經注冊的Server實例相關IP與Port信息，然后通過遍歷的方式調用Server接口通知實例此時有配置更新。
被動發現是Server實例中自主定時進行數據庫掃描，當發現新版本時通知Client端有配置變更。
1.1.2、時序圖
1.1.3、代碼位置
1.1.3.1、NotifyServiceImpl#notifyPush
當用戶在操作平臺進行配置修改時，會調用該接口進行配置變更推送，由于需要通知所有已經部署的Servers有配置更新，所以需要從注冊中心中獲取到對應的Host信息，然后通過遍歷的方式進行配置推送。
@Service
public class NotifyServiceImpl implements NotifyService, InitializingBean {
/**
* 管理平臺操作，配置變更通知
*/
@Override
public void notifyPush(final ConfigMeta meta, final long version, List<PushItemWithHostName> destinations) {
// 從注冊中心（Eureka）獲取Server實例的Hosts信息
List<String> serverUrls = getServerUrls();
if (serverUrls.isEmpty()) {
logger.warn("notify push server, {}, version: {}, but no server, {}", meta, version, destinations);
return;
}
// Server中接收變更推送的接口URL
String uri = this.notifyPushUrl;
logger.info("notify push server, {}, version: {}, uri: {}, servers: {}, {}", meta, version, uri, serverUrls, destinations);
StringBuilder sb = new StringBuilder();
for (PushItemWithHostName item : destinations) {
sb.append(item.getHostname()).append(',')
.append(item.getIp()).append(',')
.append(item.getPort()).append(Constants.LINE);
}
final String destinationsStr = sb.toString();
// 根據已注冊Server的Host列表，配置信息、配置版本等信息，執行通知推送動作
doNotify(serverUrls, uri, "push", new Function<String, Request>() {
@Override
public Request apply(String url) {
AsyncHttpClient.BoundRequestBuilder builder = getBoundRequestBuilder(url, meta, version, destinationsStr);
return builder.build();
}
});
}
/**
* 獲取注冊中心中已注冊的Server Hosts信息
*/
private List<String> getServerUrls() {
return serverListService.getOnlineServerHosts();
}
private void doNotify(List<String> serverUrls, String uri, String type, Function<String, Request> requestBuilder) {
List<ListenableFuture<Response>> futures = Lists.newArrayListWithCapacity(serverUrls.size());
for (String oneServer : serverUrls) {
String url = "http://" + oneServer + "/" + uri;
Request request = requestBuilder.apply(url);
ListenableFuture<Response> future = HttpListenableFuture.wrap(httpClient.executeRequest(request));
futures.add(future);
}
dealResult(futures, serverUrls, type);
}
}
1.1.3.2、LongPollingStoreImpl#manualPush
@Service
public class LongPollingStoreImpl implements LongPollingStore {
private static final ConcurrentMap<ConfigMeta, Cache<Listener, Listener>> listenerMappings = Maps.newConcurrentMap();
private static final int DEFAULT_THREAD_COUNT = 4;
private static final long DEFAULT_TIMEOUT = 60 * 1000L;
private static ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(
DEFAULT_THREAD_COUNT, new NamedThreadFactory("qconfig-config-listener-push"));
private static ExecutorService onChangeExecutor = Executors.newFixedThreadPool(
Runtime.getRuntime().availableProcessors(), new NamedThreadFactory("config-on-change"));
@Override
public void manualPush(ConfigMeta meta, long version, final Set<IpAndPort> ipAndPorts) {
logger.info("push client file: {}, version {}, {}", meta, version, ipAndPorts);
Set<String> ips = Sets.newHashSetWithExpectedSize(ipAndPorts.size());
for (IpAndPort ipAndPort : ipAndPorts) {
ips.add(ipAndPort.getIp());
}
manualPushIps(meta, version, ips);
}
@Override
public void manualPushIps(ConfigMeta meta, long version, final Set<String> ips) {
logger.info("push client file: {}, version {}, {}", meta, version, ips);
Stopwatch stopwatch = Stopwatch.createStarted();
try {
doChange(meta, version, Constants.PULL, new Predicate<Listener>() {
@Override
public boolean apply(Listener input) {
return ips.contains(input.getContextHolder().getIp());
}
});
} finally {
Monitor.filePushOnChangeTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
@Override
public void onChange(final ConfigMeta meta, final long version) {
logger.info("file change: {}, version {}", meta, version);
onChangeExecutor.execute(new Runnable() {
@Override
public void run() {
Stopwatch stopwatch = Stopwatch.createStarted();
try {
doChange(meta, version, Constants.UPDATE, Predicates.<Listener>alwaysTrue());
} finally {
Monitor.fileOnChangeTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
});
}
private void doChange(ConfigMeta meta, long newVersion, String type, Predicate<Listener> needChange) {
List<Listener> listeners = getListeners(meta, needChange);
if (listeners.isEmpty()) {
return;
}
Changed change = new Changed(meta, newVersion);
// 如果沒超過直接推送數量，則直接推送
if (listeners.size() <= pushConfig.getDirectPushLimit()) {
directDoChange(listeners, change, type);
} else {
// 如果超過一定數量，則scheduled定時，通過一定節奏來推送，避免驚群
PushItem pushItem = new PushItem(listeners, type, change);
scheduledExecutor.execute(new PushRunnable(pushItem));
}
}
private void directDoChange(List<Listener> listeners, Changed change, String type) {
Stopwatch stopwatch = Stopwatch.createStarted();
try {
for (Listener listener : listeners) {
logger.debug("return {}, {}", listener, change);
returnChange(change, listener, type);
}
} catch (Exception e) {
Monitor.batchReturnChangeFailCounter.inc();
logger.error("batch direct return changes error, type {}, change {}", type, change, e);
} finally {
Monitor.batchReturnChangeTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
private static class PushRunnable implements Runnable {
private final PushItem pushItem;
private PushRunnable(PushItem pushItem) {
this.pushItem = pushItem;
}
@Override
public void run() {
Stopwatch stopwatch = Stopwatch.createStarted();
try {
long start = System.currentTimeMillis();
PushConfig config = pushConfig;
int num = Math.min(pushItem.getListeners().size(), config.getPushMax());
for (int i = 0; i < num; ++i) {
Listener listener = pushItem.getListeners().poll();
returnChange(pushItem.getChange(), listener, pushItem.getType());
}
if (!pushItem.getListeners().isEmpty()) {
long elapsed = System.currentTimeMillis() - start;
long delay;
if (elapsed >= config.getPushInterval()) {
delay = 0;
} else {
delay = config.getPushInterval() - elapsed;
}
//一次推送后，以這次推送時間為起始時間，延遲一定時間后再次推送。這里的PushRunnable遞歸執行
scheduledExecutor.schedule(new PushRunnable(pushItem), delay, TimeUnit.MILLISECONDS);
}
} catch (Exception e) {
Monitor.batchReturnChangeFailCounter.inc();
logger.error("batch return changes error, {}", pushItem, e);
} finally {
Monitor.batchReturnChangeTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
}
private static void returnChange(Changed change, Listener listener, String type) {
Stopwatch stopwatch = Stopwatch.createStarted();
try {
// 通知注冊的監聽器，響應client，返回版本信息
listener.onChange(change, type);
} finally {
Monitor.returnChangeTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
}
1.2、被動推送
1.2.1、邏輯描述
首次啟動或啟動后每3分鐘，刷新一次配置的最新版本，如果出現最新版本，則觸發推送邏輯，將配置最新的版本推送至Client端中。
1.2.2、代碼位置
1.2.2.1、CacheConfigVersionServiceImpl#freshConfigVersionCache
@Service
public class CacheConfigVersionServiceImpl implements CacheConfigVersionService {
private volatile ConcurrentMap<ConfigMeta, Long> cache = Maps.newConcurrentMap();
/**
* 首次啟動或啟動后每3分鐘，刷新一次配置的最新版本
*/
@PostConstruct
public void init() {
freshConfigVersionCache();
ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
// 每3分鐘執行一次緩存刷新，判斷配置是否有最新版本
scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
Thread.currentThread().setName("fresh-config-version-thread");
try {
freshConfigVersionCache();
} catch (Throwable e) {
logger.error("fresh config version error", e);
}
}
}, 3, 3, TimeUnit.MINUTES);
}
@Override
public Optional<Long> getVersion(ConfigMeta meta) {
return Optional.fromNullable(cache.get(meta));
}
/**
* 定時刷新配置最新版本，如果出現最新版本，則觸發推送邏輯
*/
private void freshConfigVersionCache() {
Stopwatch stopwatch = Stopwatch.createStarted();
try {
logger.info("fresh config version cache");
List<VersionData<ConfigMeta>> configIds = configDao.loadAll();
ConcurrentMap<ConfigMeta, Long> newCache = new ConcurrentHashMap<ConfigMeta, Long>(configIds.size());
ConcurrentMap<ConfigMeta, Long> oldCache = this.cache;
// 判斷是否有最新版本
synchronized (this) {
for (VersionData<ConfigMeta> configId : configIds) {
long newVersion = configId.getVersion();
Long oldVersion = cache.get(configId.getData());
// 暫時不考慮delete的情況
// 從數據庫load數據先于配置更新
if (oldVersion != null && oldVersion > newVersion) {
newVersion = oldVersion;
}
// 如果有最新版本則刷新緩存
newCache.put(configId.getData(), newVersion);
}
this.cache = newCache;
}
logger.info("fresh config version cache successOf, count [{}]", configIds.size());
int updates = 0;
for (Map.Entry<ConfigMeta, Long> oldEntry : oldCache.entrySet()) {
ConfigMeta meta = oldEntry.getKey();
Long oldVersion = oldEntry.getValue();
Long newVersion = newCache.get(meta);
if (newVersion != null && newVersion > oldVersion) {
updates += 1;
// 配置變更，通知Client端
longPollingStore.onChange(meta, newVersion);
}
}
logger.info("fresh size={} config version cache from db", updates);
} finally {
Monitor.freshConfigVersionCacheTimer.update(stopwatch.elapsed().toMillis(), TimeUnit.MILLISECONDS);
}
}
}
2、變更監聽
2.1.1、邏輯描述
Client端與Server端建立長輪詢，長輪詢建立完成之后會為當前請求建立一個監聽器，當配置發生變變更時就會觸發監聽器，然后通過監聽機制結束長輪詢并返回最新的配置版本。如果沒有版本變更，長輪詢會每分鐘斷開重新建立一次。
2.1.2、時序圖
2.1.3、代碼位置
2.1.3.1、AbstractCheckVersionServlet#doPost
public abstract class AbstractCheckVersionServlet extends AbstractServlet {
private static final long serialVersionUID = -8278568383506314625L;
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
...
checkVersion(requests, req, resp);
}
}
2.1.3.2、LongPollingCheckServlet#checkVersion
public class LongPollingCheckServlet extends AbstractCheckVersionServlet {
@Override
protected void checkVersion(List<CheckRequest> checkRequests, HttpServletRequest req, HttpServletResponse resp)
throws ServletException, IOException {
...
try {
// 異步
AsyncContext context = req.startAsync();
// (核心流程，重點關注)，執行版本檢查（長輪詢）
getLongPollingProcessService().process(context, checkRequests);
} catch (Throwable e) {
// never come here !!!
logger.error("服務異常", e);
}
}
}
2.1.3.3、LongPollingProcessServiceImpl#process
@Service
public class LongPollingProcessServiceImpl implements LongPollingProcessService {
@PostConstruct
public void init() {
MapConfig config = MapConfig.get("config.properties");
config.asMap();
// 向config中添加監聽器
config.addListener(new Configuration.ConfigListener<Map<String, String>>() {
@Override
public void onLoad(Map<String, String> conf) {
String newTimeout = conf.get("longPolling.server.timeout");
if (!Strings.isNullOrEmpty(newTimeout)) {
timeout = Numbers.toLong(newTimeout, DEFAULT_TIMEOUT);
}
}
});
}
// 核心邏輯，重點關注
@Override
public void process(AsyncContext context, List<CheckRequest> requests) {
IpAndPort address = new IpAndPort(clientInfoService.getIp(), clientInfoService.getPort());
AsyncContextHolder contextHolder = new AsyncContextHolder(context, address);
// 設置超時
context.setTimeout(timeout);
// 設置監聽器
context.addListener(new TimeoutServletListener(contextHolder));
processCheckRequests(requests, clientInfoService.getIp(), contextHolder);
}
private void processCheckRequests(List<CheckRequest> requests, String ip, AsyncContextHolder contextHolder) {
CheckResult result = checkService.check(requests, ip, qFileFactory);
logger.info("profile:{}, result change list {} for check request {}", clientInfoService.getProfile(), result.getChanges(), requests);
if (!result.getChanges().isEmpty()) {
returnChanges(AbstractCheckConfigServlet.formatOutput(CheckUtil.processStringCase(result.getChanges())), contextHolder, Constants.UPDATE);
return;
}
// 為該請求注冊監聽器，并存放至longPollingStore中
addListener(result.getRequestsNoChange(), contextHolder);
// 注冊client
registerOnlineClients(result, contextHolder);
}
private void addListener(Map<CheckRequest, QFile> requests, AsyncContextHolder contextHolder) {
for (Map.Entry<CheckRequest, QFile> noChangeEntry : requests.entrySet()) {
CheckRequest request = noChangeEntry.getKey();
QFile qFile = noChangeEntry.getValue();
if (!contextHolder.isComplete()) {
// 根據請求創建監聽器
Listener listener = qFile.createListener(request, contextHolder);
// 將監聽器存儲至longPollingStore
longPollingStore.addListener(listener);
}
}
}
private void registerOnlineClients(CheckResult result, AsyncContextHolder contextHolder) {
Map<CheckRequest, QFile> noChanges = Maps.newHashMapWithExpectedSize(
result.getRequestsNoChange().size() + result.getRequestsLockByFixVersion().size());
noChanges.putAll(result.getRequestsNoChange());
noChanges.putAll(result.getRequestsLockByFixVersion());
for (Map.Entry<CheckRequest, QFile> noChangeEntry : noChanges.entrySet()) {
CheckRequest request = noChangeEntry.getKey();
QFile qFile = noChangeEntry.getValue();
if (!contextHolder.isComplete()) {
long version = request.getVersion();
ConfigMeta meta = qFile.getRealMeta();
String ip = contextHolder.getIp();
if (qFile instanceof InheritQFileV2) {
InheritQFileV2 inheritQFile = (InheritQFileV2) qFile;
Optional<Long> optional = inheritQFile.getCacheConfigInfoService().getVersion(inheritQFile.getRealMeta());
version = optional.isPresent() ? optional.get() : version;
onlineClientListService.register(inheritQFile.getRealMeta(), ip, version);
} else {
// 注冊client，admin（管理平臺）獲取已經連接的client信息，其中包括ip、配置版本
onlineClientListService.register(meta, ip, version);
}
}
}
}
/**
* 配置變化，執行返回
*/
private void returnChanges(String change, AsyncContextHolder contextHolder, String type) {
contextHolder.completeRequest(new ChangeReturnAction(change, type));
}
}
2.1.3.4、CheckService#check
@Service
public class CheckServiceImpl implements CheckService {
...
@Override
public CheckResult check(List<CheckRequest> requests, String ip, QFileFactory qFileFactory) {
List<CheckRequest> requestsNoFile = Lists.newArrayList();
Map<CheckRequest, Changed> changes = Maps.newHashMap();
Map<CheckRequest, QFile> requestNoChange = Maps.newHashMap();
Map<CheckRequest, QFile> requestsLockByFixVersion = Maps.newHashMap();
for (CheckRequest request : requests) {
ConfigMeta meta = new ConfigMeta(request.getGroup(), request.getDataId(), request.getProfile());
Optional<QFile> qFileOptional = qFileFactory.create(meta, cacheConfigInfoService);
if (!qFileOptional.isPresent()) {
requestsNoFile.add(request);
continue;
}
QFile qFile = qFileOptional.get();
// 核心邏輯，檢測版本
Optional<Changed> changedOptional = qFile.checkChange(request, ip);
if (changedOptional.isPresent()) {
Optional<Changed> resultChange = repairChangeWithFixVersion(qFile, request, ip, changedOptional.get());
if (resultChange.isPresent()) {
changes.put(request, resultChange.get());
} else {
requestsLockByFixVersion.put(request, qFile);
}
} else {
requestNoChange.put(request, qFile);
}
}
return new CheckResult(requestsNoFile, changes, requestNoChange, requestsLockByFixVersion);
}
}
2.1.3.5、QFileEntityV1#checkChange
public class QFileEntityV1 extends AbstractQFileEntity implements QFile {
public QFileEntityV1(ConfigMeta meta,
CacheConfigInfoService cacheConfigInfoService,
ConfigStore configStore,
LogService logService,
ClientInfoService clientInfoService) {
super(meta, cacheConfigInfoService, configStore, logService, clientInfoService);
}
@Override
public Optional<Changed> checkChange(CheckRequest request, String ip) {
ConfigMeta meta = getSourceMeta();
// 從緩存中獲取配置文件的最新版本
Optional<Long> version = getCacheConfigInfoService().getVersion(meta, ip);
if (!version.isPresent()) {
return Optional.absent();
}
if (version.get() <= request.getVersion()) {
return Optional.absent();
}
return Optional.of(new Changed(meta.getGroup(), meta.getDataId(), meta.getProfile(), version.get()));
}
}
2.1.3.6、CacheConfigInfoService#getVersion
@Service("cacheConfigInfoService")
public class CacheConfigInfoService implements ConfigInfoService {
...
@Override
public Optional<Long> getVersion(ConfigMeta meta, String ip) {
// 獲取配置已發布的最新版本
Optional<Long> publishVersion = getVersion(meta);
// 獲取推送給該IP的配置的最新灰度版本
Optional<Long> pushVersion = getPushVersion(meta, ip);
return VersionUtil.getLoadVersion(publishVersion, pushVersion);
}
}
3、Client配置拉取
3.1.1、邏輯描述
根據長輪詢后Client端獲取到的配置文件對應的最新版本信息，查詢最新的配置數據。查詢順序是先查詢緩存，如果查找不到則通過本地文件查找，如果再查不到則查詢數據庫。這樣可以有效緩解數據庫壓力。
3.1.2、代碼位置
3.1.2.1、ConfigStoreImpl#findConfig
@Service
public class ConfigStoreImpl implements ConfigStore {
private LoadingCache<VersionData<ConfigMeta>, ChecksumData<String>> configCache;
@PostConstruct
private void init() {
configCache = CacheBuilder.newBuilder()
.maximumSize(5000) // 最大數量
.expireAfterAccess(10, TimeUnit.SECONDS) // 訪問失效時間
.recordStats()
.build(new CacheLoader<VersionData<ConfigMeta>, ChecksumData<String>>() {
@Override
public ChecksumData<String> load(VersionData<ConfigMeta> configId) throws ConfigNotFoundException {
return loadConfig(configId);
}
});
Metrics.gauge("configFile_notFound_cache_hitRate", new Supplier<Double>() {
@Override
public Double get() {
return configCache.stats().hitRate();
}
});
}
/**
* 查本地guava cache
*/
@Override
public ChecksumData<String> findConfig(VersionData<ConfigMeta> configId) throws ConfigNotFoundException {
try {
return configCache.get(configId);
} catch (ExecutionException e) {
if (e.getCause() instanceof ConfigNotFoundException) {
throw (ConfigNotFoundException) e.getCause();
} else {
log.error("find config error, configId:{}", configId, e);
throw new RuntimeException(e.getCause());
}
}
}
/**
* 從本地文件或數據庫中獲取配置信息
*/
private ChecksumData<String> loadConfig(VersionData<ConfigMeta> configId) throws ConfigNotFoundException {
// 從本地配置文件中查詢配置信息
ChecksumData<String> config = findFromDisk(configId);
if (config != null) {
return config;
}
String groupId = configId.getData().getGroup();
Monitor.notFoundConfigFileFromDiskCounterInc(groupId);
log.warn("config not found from disk: {}", configId);
// 從數據庫中加載配置數據
config = findFromDb(configId);
if (config != null) {
return config;
}
Monitor.notFoundConfigFileFromDbCounterInc(groupId);
throw new ConfigNotFoundException();
}
private ChecksumData<String> findFromDb(VersionData<ConfigMeta> configId) {
ChecksumData<String> config = configDao.loadFromCandidateSnapshot(configId);
if (config != null) {
saveToFile(configId, config);
}
return config;
}
}
三、最后
《碼頭工人的一千零一夜》是一位專注于技術干貨分享的博主，追隨博主的文章，你將深入了解業界最新的技術趨勢，以及在Java開發和安全領域的實用經驗分享。無論你是開發人員還是對逆向工程感興趣的愛好者，都能在《碼頭工人的一千零一夜》找到有價值的知識和見解。
懂得不多，做得太少。歡迎批評、指正。
浙公網安備 33010602011771號