【架構師系列】Apollo配置中心之Server端（ConfigSevice）（三）
聲明
原創文章，轉載請標注。http://www.rzrgm.cn/boycelee/p/17993697
《碼頭工人的一千零一夜》是一位專注于技術干貨分享的博主，追隨博主的文章，你將深入了解業界最新的技術趨勢，以及在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
一、通知機制
二、架構思考
1、配置變更如何通知客戶端？
（1）如何建立長輪詢？
2、客戶端如何拉取數據？
（1）如何拉取數據？
3、如何發現變更數據？
（1）為什么使用Config Service定時掃描ReleaseMessage的方式？
（2）為什么不采用Client調用Config Service直接查詢的方式？
三、源碼剖析
1、配置監聽
1.1、建立長輪詢
1.1.1、邏輯描述
1.1.2、時序圖
1.1.3、代碼位置
1.1.3.1、NotificationControllerV2#pollNotification
@RestController
@RequestMapping("/notifications/v2")
public class NotificationControllerV2 implements ReleaseMessageListener {
...
private final Multimap<String, DeferredResultWrapper> deferredResults = Multimaps.synchronizedSetMultimap(TreeMultimap.create(String.CASE_INSENSITIVE_ORDER, Ordering.natural()));
@GetMapping
public DeferredResult<ResponseEntity<List<ApolloConfigNotification>>> pollNotification(
@RequestParam(value = "appId") String appId,
@RequestParam(value = "cluster") String cluster,
@RequestParam(value = "notifications") String notificationsAsString,
@RequestParam(value = "dataCenter", required = false) String dataCenter,
@RequestParam(value = "ip", required = false) String clientIp) {
List<ApolloConfigNotification> notifications = null;
// 反序列化
try {
notifications =
gson.fromJson(notificationsAsString, notificationsTypeReference);
} catch (Throwable ex) {
Tracer.logError(ex);
}
// (非核心，不關注)
Map<String, ApolloConfigNotification> filteredNotifications = filterNotifications(appId, notifications);
// (核心流程，重點關注)
// 使用Wrapper封裝DeferredResult，利用Spring的DeferredResult + tomcat實現長輪詢。
DeferredResultWrapper deferredResultWrapper = new DeferredResultWrapper(bizConfig.longPollingTimeoutInMilli());
Set<String> namespaces = Sets.newHashSetWithExpectedSize(filteredNotifications.size());
Map<String, Long> clientSideNotifications = Maps.newHashMapWithExpectedSize(filteredNotifications.size());
for (Map.Entry<String, ApolloConfigNotification> notificationEntry : filteredNotifications.entrySet()) {
String normalizedNamespace = notificationEntry.getKey();
ApolloConfigNotification notification = notificationEntry.getValue();
namespaces.add(normalizedNamespace);
clientSideNotifications.put(normalizedNamespace, notification.getNotificationId());
if (!Objects.equals(notification.getNamespaceName(), normalizedNamespace)) {
// namespace名的關系映射（非核心，不關注）
deferredResultWrapper.recordNamespaceNameNormalizedResult(notification.getNamespaceName(), normalizedNamespace);
}
}
// watchedKeysMap 格式: namespace : appId_cluster_namespace
Multimap<String, String> watchedKeysMap = watchKeysUtil.assembleAllWatchKeys(appId, cluster, namespaces, dataCenter);
Set<String> watchedKeys = Sets.newHashSet(watchedKeysMap.values());
/**
* 1、set deferredResult before the check, for avoid more waiting
* If the check before setting deferredResult,it may receive a notification the next time
* when method handleMessage is executed between check and set deferredResult.
*/
deferredResultWrapper
.onTimeout(() -> logWatchedKeys(watchedKeys, "Apollo.LongPoll.TimeOutKeys"));
// 完成時執行，將對應deferredResultWrapper從deferredResults中移除，表示該本次長輪詢結束
deferredResultWrapper.onCompletion(() -> {
//unregister all keys
for (String key : watchedKeys) {
deferredResults.remove(key, deferredResultWrapper);
}
logWatchedKeys(watchedKeys, "Apollo.LongPoll.CompletedKeys");
});
// (核心流程，重點關注)
// watchedKey : 格式：appId_cluster_namespace
// register all keys
// 將namespace:deferredResult注冊至deferredResults（Map容器）中。
// 多個namespace對應同一個deferredResult。當namespace發生變化時，就會從deferredResults找到對應的deferredResult，通知客戶端。
// 思考：DeferredResult是什么？和長輪詢有什么關系？可以和其他的異步工具有什么區別？
for (String key : watchedKeys) {
this.deferredResults.put(key, deferredResultWrapper);
}
/**
* 2、check new release
*/
// (核心流程，重點關注)
// 同步檢測是否有最新版本，通過WatchedKeys（格式：appId_cluster_namespace）拉取最新通知信息（如果有變更直接返回，并不會等后續通知）。
List<ReleaseMessage> latestReleaseMessages = releaseMessageService.findLatestReleaseMessagesGroupByMessages(watchedKeys);
/**
* Manually close the entity manager.
* Since for async request, Spring won't do so until the request is finished,
* which is unacceptable since we are doing long polling - means the db connection would be hold
* for a very long time
*/
entityManagerUtil.closeEntityManager();
// (核心流程，重點關注)
// 對latestReleaseMessages進行封裝，將其封裝成ApolloConfigNotification類型
// 此處ApolloConfigNotification中只返回配置發生變更的namespace及其對應的notificationId
List<ApolloConfigNotification> newNotifications = getApolloConfigNotifications(namespaces, clientSideNotifications, watchedKeysMap, latestReleaseMessages);
// (核心流程，重點關注)
// 注意：這里是同步返回，如果有查詢發現有最新版本，直接返回，不需要等待通知。
// DeferredResult則需要用戶在代碼中手動set值到DeferredResult，否則即便異步線程中的任務執行完畢，DeferredResult仍然不會向客戶端返回任何結果。
// 如果是有新配置，則通過handleMessage函數向deferredResultWrapper#setResult賦值.
if (!CollectionUtils.isEmpty(newNotifications)) {
// 非主動變更，其他情況通過此處進行相應。
deferredResultWrapper.setResult(newNotifications);
}
return deferredResultWrapper.getResult();
}
/**
* 此處ApolloConfigNotification中只返回配置發生變更的namespace及其對應的notificationId
**/
private List<ApolloConfigNotification> getApolloConfigNotifications(Set<String> namespaces,
Map<String, Long> clientSideNotifications,
Multimap<String, String> watchedKeysMap,
List<ReleaseMessage> latestReleaseMessages) {
List<ApolloConfigNotification> newNotifications = Lists.newArrayList();
// 判斷是否查詢到namespace的最新版本消息
if (!CollectionUtils.isEmpty(latestReleaseMessages)) {
Map<String, Long> latestNotifications = Maps.newHashMap();
for (ReleaseMessage releaseMessage : latestReleaseMessages) {
latestNotifications.put(releaseMessage.getMessage(), releaseMessage.getId());
}
// 遍歷namespace
for (String namespace : namespaces) {
long clientSideId = clientSideNotifications.get(namespace);
long latestId = ConfigConsts.NOTIFICATION_ID_PLACEHOLDER;
Collection<String> namespaceWatchedKeys = watchedKeysMap.get(namespace);
for (String namespaceWatchedKey : namespaceWatchedKeys) {
// 獲取最新版本的namespace對應的nofiticationId
long namespaceNotificationId =
latestNotifications.getOrDefault(namespaceWatchedKey, ConfigConsts.NOTIFICATION_ID_PLACEHOLDER);
if (namespaceNotificationId > latestId) {
latestId = namespaceNotificationId;
}
}
// 如果Config Service中的namespace對應的通知編號大于Client上傳的namespace對應的通知編號，則說明有配置變更，就執行封裝動作，將最新的namespace對應的通知編號（notificationId）返回
if (latestId > clientSideId) {
ApolloConfigNotification notification = new ApolloConfigNotification(namespace, latestId);
namespaceWatchedKeys.stream().filter(latestNotifications::containsKey).forEach(namespaceWatchedKey ->
notification.addMessage(namespaceWatchedKey, latestNotifications.get(namespaceWatchedKey)));
newNotifications.add(notification);
}
}
}
return newNotifications;
}
}
1.1.3.2、ReleaseMessageServiceWithCache#findLatestReleaseMessagesGroupByMessages
@Service
public class ReleaseMessageServiceWithCache implements ReleaseMessageListener, InitializingBean {
private ConcurrentMap<String, ReleaseMessage> releaseMessageCache;
...
public List<ReleaseMessage> findLatestReleaseMessagesGroupByMessages(Set<String> messages) {
// messages格式：appId_cluster_namespace
if (CollectionUtils.isEmpty(messages)) {
return Collections.emptyList();
}
List<ReleaseMessage> releaseMessages = Lists.newArrayList();
// 此處的message命名為namespaces更合適
for (String message : messages) {
// 獲取緩存中namespace的版本信息
ReleaseMessage releaseMessage = releaseMessageCache.get(message);
if (releaseMessage != null) {
releaseMessages.add(releaseMessage);
}
}
return releaseMessages;
}
...
}
1.2、刷新ReleaseMessage緩存
1.3.1、邏輯描述
更新 ReleaseMessages，管理 releaseMessageCache，其中鍵為 appId_cluster_namespace，值為通知編號 notificationId。
1.3.2、代碼位置
1.3.2.1、ReleaseMessageServiceWithCache#afterPropertiesSet
@Service
public class ReleaseMessageServiceWithCache implements ReleaseMessageListener, InitializingBean {
// 維護空間最新的，結構為appId_cluster_namespace : notificationId
private ConcurrentMap<String, ReleaseMessage> releaseMessageCache;
private AtomicBoolean doScan;
private ExecutorService executorService;
private void initialize() {
releaseMessageCache = Maps.newConcurrentMap();
doScan = new AtomicBoolean(true);
// 此處初始化的是單線程的線程池，不帶定時任務。
executorService = Executors.newSingleThreadExecutor(ApolloThreadFactory
.create("ReleaseMessageServiceWithCache", true));
}
public List<ReleaseMessage> findLatestReleaseMessagesGroupByMessages(Set<String> messages) {
// messages格式：appId_cluster_namespace
if (CollectionUtils.isEmpty(messages)) {
return Collections.emptyList();
}
List<ReleaseMessage> releaseMessages = Lists.newArrayList();
for (String message : messages) {
ReleaseMessage releaseMessage = releaseMessageCache.get(message);
if (releaseMessage != null) {
releaseMessages.add(releaseMessage);
}
}
return releaseMessages;
}
@Override
public void handleMessage(ReleaseMessage message, String channel) {
//Could stop once the ReleaseMessageScanner starts to work
doScan.set(false);
logger.info("message received - channel: {}, message: {}", channel, message);
String content = message.getMessage();
Tracer.logEvent("Apollo.ReleaseMessageService.UpdateCache", String.valueOf(message.getId()));
if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(content)) {
return;
}
// 計算本地ReleaseMessageId與觸發handleMessage的ReleaseMessageId的gap
long gap = message.getId() - maxIdScanned;
// 如果gap等于1，直接合并
if (gap == 1) {
mergeReleaseMessage(message);
} else if (gap > 1) {
//gap found!
// 如果gap大于1，加載gap間缺失的ReleaseMessage查詢出來，并將新查詢出的ReleaseMessages與歷史數據進行比較，releaseMessageCache中維護最新的ReleaseMessageId（notificationId）
loadReleaseMessages(maxIdScanned);
}
}
@Override
public void afterPropertiesSet() throws Exception {
// 讀取配置
populateDataBaseInterval();
//block the startup process until load finished
//this should happen before ReleaseMessageScanner due to autowire
// 初始化，拉取ReleaseMessages
loadReleaseMessages(0);
// 異步拉取增量ReleaseMessages。（注意：這里executorService不是定時任務，而是單線程的線程池）
// 目的：處理初始化時拉取ReleaseMessages產生的遺漏問題
// 這里可以理解為fix（可以不關注）
executorService.submit(() -> {
while (doScan.get() && !Thread.currentThread().isInterrupted()) {
Transaction transaction = Tracer.newTransaction("Apollo.ReleaseMessageServiceWithCache",
"scanNewReleaseMessages");
try {
// 加載ReleaseMessages
loadReleaseMessages(maxIdScanned);
transaction.setStatus(Transaction.SUCCESS);
} catch (Throwable ex) {
transaction.setStatus(ex);
logger.error("Scan new release messages failed", ex);
} finally {
transaction.complete();
}
try {
scanIntervalTimeUnit.sleep(scanInterval);
} catch (InterruptedException e) {
//ignore
}
}
});
}
/**
*
* @param releaseMessage
*/
private synchronized void mergeReleaseMessage(ReleaseMessage releaseMessage) {
ReleaseMessage old = releaseMessageCache.get(releaseMessage.getMessage());
// 判斷當前ReleaseMessages的id是否大于歷史ReleaseMessages的id，如果大于則更新緩存
if (old == null || releaseMessage.getId() > old.getId()) {
// message 內容為： appId_cluster_namespace
releaseMessageCache.put(releaseMessage.getMessage(), releaseMessage);
maxIdScanned = releaseMessage.getId();
}
}
private void loadReleaseMessages(long startId) {
boolean hasMore = true;
while (hasMore && !Thread.currentThread().isInterrupted()) {
//current batch is 500
// 此處邏輯和AppNamespaceServiceWithCache的一樣
// 批量獲取大于startId的500條ReleaseMessages數據（返回升序）
// 思考：需要掃描才能知道最新的消息ID，這樣的設計不太好
List<ReleaseMessage> releaseMessages = releaseMessageRepository
.findFirst500ByIdGreaterThanOrderByIdAsc(startId);
if (CollectionUtils.isEmpty(releaseMessages)) {
break;
}
// 將新查詢出的ReleaseMessages與歷史數據進行比較，releaseMessageCache中維護最新的ReleaseMessageId（notificationId）
releaseMessages.forEach(this::mergeReleaseMessage);
// 獲取新的startId，作為當前的最新數據標記，便于后續在此startId基礎上拉取后續新的ReleaseMessages
int scanned = releaseMessages.size();
startId = releaseMessages.get(scanned - 1).getId();
// 當拉取數據（scanned）大于500時，說明后續還有數據，則繼續執行進入while中，否則退出
hasMore = scanned == 500;
logger.info("Loaded {} release messages with startId {}", scanned, startId);
}
}
2、變更推送
Admin Service將發布后的配置，通過消息的方式發送給Config Service，然后Config Service通知對應的Client。此處可以通過消息中間件來實現消息的生產與發現，但考慮到一個中間件的引入的同時也會帶來很多不確定性隱患，所以通過數據庫的方式實現消息的生產與消費。
2.1、觸發變更
2.1.1、邏輯描述
用戶的操作發布后，通過AdminService向數據庫中的ReleaseMessage表插入配置變更通知編號。
2.1.2、代碼位置
2.1.3.1、DatabaseMessageSender#sendMessage
@Component
public class DatabaseMessageSender implements MessageSender {
private BlockingQueue<Long> toClean = Queues.newLinkedBlockingQueue(CLEAN_QUEUE_MAX_SIZE);
private final ExecutorService cleanExecutorService;
private final ReleaseMessageRepository releaseMessageRepository;
public DatabaseMessageSender(final ReleaseMessageRepository releaseMessageRepository) {
cleanExecutorService = Executors.newSingleThreadExecutor(ApolloThreadFactory.create("DatabaseMessageSender", true));
cleanStopped = new AtomicBoolean(false);
this.releaseMessageRepository = releaseMessageRepository;
}
@Override
@Transactional
public void sendMessage(String message, String channel) {
logger.info("Sending message {} to channel {}", message, channel);
// 只發布APOLLO_RELEASE_TOPIC的數據
if (!Objects.equals(channel, Topics.APOLLO_RELEASE_TOPIC)) {
logger.warn("Channel {} not supported by DatabaseMessageSender!", channel);
return;
}
Tracer.logEvent("Apollo.AdminService.ReleaseMessage", message);
Transaction transaction = Tracer.newTransaction("Apollo.AdminService", "sendMessage");
try {
// 存儲 ReleaseMessage
ReleaseMessage newMessage = releaseMessageRepository.save(new ReleaseMessage(message));
// 添加到阻塞阻塞隊列中，成功失敗都會立即返回（會在初始化時，開啟一個線程處理toClean）
toClean.offer(newMessage.getId());
transaction.setStatus(Transaction.SUCCESS);
} catch (Throwable ex) {
logger.error("Sending message to database failed", ex);
transaction.setStatus(ex);
throw ex;
} finally {
transaction.complete();
}
}
...
}
2.1.3.2、ReleaseMessage
@Entity
@Table(name = "ReleaseMessage")
public class ReleaseMessage {
// 自增ID
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Column(name = "Id")
private long id;
// message 內容為： appId_cluster_namespace
@Column(name = "Message", nullable = false)
private String message;
// 更新時間
@Column(name = "DataChange_LastTime")
private Date dataChangeLastModifiedTime;
}
2.2、感知變更
2.2.1、邏輯描述
在bean初始化結束后執行ReleaseMessageScanner#afterPropertiesSet函數中的操作，定時掃描數據庫，獲取最大的ReleaseMessageId。一旦有新的ReleaseMessage就會立即通過fireMessageScanned通知監聽器。
2.2.2、時序圖
2.2.3、代碼位置
2.2.3.1、ReleaseMessageScanner#afterPropertiesSet
public class ReleaseMessageScanner implements InitializingBean {
@Autowired
private ReleaseMessageRepository releaseMessageRepository;
private int databaseScanInterval;
private final List<ReleaseMessageListener> listeners;
private final ScheduledExecutorService executorService;
private final Map<Long, Integer> missingReleaseMessages; // missing release message id => age counter
private long maxIdScanned;
public ReleaseMessageScanner() {
listeners = Lists.newCopyOnWriteArrayList();
executorService = Executors.newScheduledThreadPool(1, ApolloThreadFactory
.create("ReleaseMessageScanner", true));
missingReleaseMessages = Maps.newHashMap();
}
@Override
public void afterPropertiesSet() throws Exception {
databaseScanInterval = bizConfig.releaseMessageScanIntervalInMilli();
// 獲取最大的ReleaseMessageId
maxIdScanned = loadLargestMessageId();
executorService.scheduleWithFixedDelay(() -> {
Transaction transaction = Tracer.newTransaction("Apollo.ReleaseMessageScanner", "scanMessage");
try {
scanMissingMessages();
// 掃描數據庫中的ReleaseMessage
scanMessages();
transaction.setStatus(Transaction.SUCCESS);
} catch (Throwable ex) {
transaction.setStatus(ex);
logger.error("Scan and send message failed", ex);
} finally {
transaction.complete();
}
}, databaseScanInterval, databaseScanInterval, TimeUnit.MILLISECONDS);
}
/**
* Scan messages, continue scanning until there is no more messages
*/
private void scanMessages() {
boolean hasMoreMessages = true;
while (hasMoreMessages && !Thread.currentThread().isInterrupted()) {
// 循環獲取最新ReleaseMessage
hasMoreMessages = scanAndSendMessages();
}
}
/**
* scan messages and send
*
* @return whether there are more messages
*/
private boolean scanAndSendMessages() {
// 批量獲取500條ReleaseMessage數據（升序）
//current batch is 500
List<ReleaseMessage> releaseMessages =
releaseMessageRepository.findFirst500ByIdGreaterThanOrderByIdAsc(maxIdScanned);
if (CollectionUtils.isEmpty(releaseMessages)) {
return false;
}
// 通知Listener，觸發監聽器
fireMessageScanned(releaseMessages);
int messageScanned = releaseMessages.size();
long newMaxIdScanned = releaseMessages.get(messageScanned - 1).getId();
// check id gaps, possible reasons are release message not committed yet or already rolled back
if (newMaxIdScanned - maxIdScanned > messageScanned) {
recordMissingReleaseMessageIds(releaseMessages, maxIdScanned);
}
maxIdScanned = newMaxIdScanned;
// 一次拉取500條數據，如果超過500條則分配多次拉取
return messageScanned == 500;
}
private void scanMissingMessages() {
Set<Long> missingReleaseMessageIds = missingReleaseMessages.keySet();
Iterable<ReleaseMessage> releaseMessages = releaseMessageRepository
.findAllById(missingReleaseMessageIds);
fireMessageScanned(releaseMessages);
releaseMessages.forEach(releaseMessage -> {
missingReleaseMessageIds.remove(releaseMessage.getId());
});
growAndCleanMissingMessages();
}
/**
* Notify listeners with messages loaded
* @param messages
*/
private void fireMessageScanned(Iterable<ReleaseMessage> messages) {
for (ReleaseMessage message : messages) {
for (ReleaseMessageListener listener : listeners) {
try {
// 通知Listener，觸發監聽器
listener.handleMessage(message, Topics.APOLLO_RELEASE_TOPIC);
} catch (Throwable ex) {
Tracer.logError(ex);
logger.error("Failed to invoke message listener {}", listener.getClass(), ex);
}
}
}
}
}
2.3、推送變更
2.3.1、邏輯描述
向與ConfigSerivce建立監聽長輪詢的Client端推送變更的配置信息，為了避免“驚群效應”出現，會使用線程池分批進行消息推送。
2.3.3、代碼位置
2.3.3.1、NotificationControllerV2#handleMessage
@RestController
@RequestMapping("/notifications/v2")
public class NotificationControllerV2 implements ReleaseMessageListener {
@Override
public void handleMessage(ReleaseMessage message, String channel) {
logger.info("message received - channel: {}, message: {}", channel, message);
String content = message.getMessage();
Tracer.logEvent("Apollo.LongPoll.Messages", content);
if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(content)) {
return;
}
// 獲取對應namespace
String changedNamespace = retrieveNamespaceFromReleaseMessage.apply(content);
if (Strings.isNullOrEmpty(changedNamespace)) {
logger.error("message format invalid - {}", content);
return;
}
if (!deferredResults.containsKey(content)) {
return;
}
//create a new list to avoid ConcurrentModificationException
List<DeferredResultWrapper> results = Lists.newArrayList(deferredResults.get(content));
ApolloConfigNotification configNotification = new ApolloConfigNotification(changedNamespace, message.getId());
configNotification.addMessage(content, message.getId());
//do async notification if too many clients
// 使用線程池分批進行消息推送，避免“驚群效應”出現
if (results.size() > bizConfig.releaseMessageNotificationBatch()) {
largeNotificationBatchExecutorService.submit(() -> {
logger.debug("Async notify {} clients for key {} with batch {}", results.size(), content,
bizConfig.releaseMessageNotificationBatch());
for (int i = 0; i < results.size(); i++) {
if (i > 0 && i % bizConfig.releaseMessageNotificationBatch() == 0) {
try {
TimeUnit.MILLISECONDS.sleep(bizConfig.releaseMessageNotificationBatchIntervalInMilli());
} catch (InterruptedException e) {
//ignore
}
}
logger.debug("Async notify {}", results.get(i));
results.get(i).setResult(configNotification);
}
});
return;
}
logger.debug("Notify {} clients for key {}", results.size(), content);
// 將變更消息設置進DeferredResult中
for (DeferredResultWrapper result : results) {
result.setResult(configNotification);
}
logger.debug("Notification completed");
}
}
3、配置拉取
3.1、構建緩存
3.1.1、邏輯描述
為了降低數據庫的查詢壓力，會將熱點數據緩存至GuavaCache中。ConfigServiceWithCache會構建兩個緩存，分別是configCache和configIdCache，其中configCache可以通過空間名稱查詢具體配置，configIdCache可以通過通知編號（notificationId）查詢具體配置。
3.1.2、代碼位置
3.1.2.1、ConfigServiceWithCache#initialize
public class ConfigServiceWithCache extends AbstractConfigService {
...
private LoadingCache<String, ConfigCacheEntry> configCache;
private LoadingCache<Long, Optional<Release>> configIdCache;
/**
* 初始化配置加載
*/
@PostConstruct
void initialize() {
// key為namespace，value為通知編號(notificationId)及其對應的配置信息(Release)
// 訪問后配置緩存有效時間為60分鐘
configCache = CacheBuilder.newBuilder()
.expireAfterAccess(DEFAULT_EXPIRED_AFTER_ACCESS_IN_MINUTES, TimeUnit.MINUTES)
.build(new CacheLoader<String, ConfigCacheEntry>() {
@Override
public ConfigCacheEntry load(String key) throws Exception {
List<String> namespaceInfo = STRING_SPLITTER.splitToList(key);
if (namespaceInfo.size() != 3) {
Tracer.logError(
new IllegalArgumentException(String.format("Invalid cache load key %s", key)));
return nullConfigCacheEntry;
}
Transaction transaction = Tracer.newTransaction(TRACER_EVENT_CACHE_LOAD, key);
try {
// 加載ReleaseMessage對象，具體就是通知編號notificationId，ReleaseMessage中包含參數有id、message（內容：appId_cluster_namespace）、dataChangeLastModifiedTime
ReleaseMessage latestReleaseMessage = releaseMessageService.findLatestReleaseMessageForMessages(Lists
.newArrayList(key));
// 獲取Release 最新配置信息
Release latestRelease = releaseService.findLatestActiveRelease(namespaceInfo.get(0), namespaceInfo.get(1),
namespaceInfo.get(2));
transaction.setStatus(Transaction.SUCCESS);
// 獲取通知編號
long notificationId = latestReleaseMessage == null ? ConfigConsts.NOTIFICATION_ID_PLACEHOLDER : latestReleaseMessage
.getId();
if (notificationId == ConfigConsts.NOTIFICATION_ID_PLACEHOLDER && latestRelease == null) {
return nullConfigCacheEntry;
}
// 緩存key為通知編號(notificationId),value為對應的具體配置信息(Release)
return new ConfigCacheEntry(notificationId, latestRelease);
} catch (Throwable ex) {
transaction.setStatus(ex);
throw ex;
} finally {
transaction.complete();
}
}
});
// key為通知編號(notificationId)，value為具體配置信息
configIdCache = CacheBuilder.newBuilder()
.expireAfterAccess(DEFAULT_EXPIRED_AFTER_ACCESS_IN_MINUTES, TimeUnit.MINUTES)
.build(new CacheLoader<Long, Optional<Release>>() {
@Override
public Optional<Release> load(Long key) throws Exception {
Transaction transaction = Tracer.newTransaction(TRACER_EVENT_CACHE_LOAD_ID, String.valueOf(key));
try {
// key為notificationId，value為具體配置
Release release = releaseService.findActiveOne(key);
transaction.setStatus(Transaction.SUCCESS);
return Optional.ofNullable(release);
} catch (Throwable ex) {
transaction.setStatus(ex);
throw ex;
} finally {
transaction.complete();
}
}
});
}
}
3.2、查詢配置
3.2.1、邏輯描述
通過前序長輪詢流程通知，獲取的namespace對應的最新通知編號(notificationId)，來查詢最新配置。
3.2.2、時序圖
3.2.3、代碼位置
3.2.3.1、ConfigController#queryConfig
public class ConfigController {
private final ConfigService configService;
private final AppNamespaceServiceWithCache appNamespaceService;
...
@GetMapping(value = "/{appId}/{clusterName}/{namespace:.+}")
public ApolloConfig queryConfig(@PathVariable String appId, @PathVariable String clusterName,
@PathVariable String namespace,
@RequestParam(value = "dataCenter", required = false) String dataCenter,
@RequestParam(value = "releaseKey", defaultValue = "-1") String clientSideReleaseKey,
@RequestParam(value = "ip", required = false) String clientIp,
@RequestParam(value = "messages", required = false) String messagesAsString,
HttpServletRequest request, HttpServletResponse response) throws IOException {
String originalNamespace = namespace;
//strip out .properties suffix
namespace = namespaceUtil.filterNamespaceName(namespace);
//fix the character case issue, such as FX.apollo <-> fx.apollo
namespace = namespaceUtil.normalizeNamespace(appId, namespace);
if (Strings.isNullOrEmpty(clientIp)) {
clientIp = tryToGetClientIp(request);
}
// 反序列化
ApolloNotificationMessages clientMessages = transformMessages(messagesAsString);
List<Release> releases = Lists.newLinkedList();
String appClusterNameLoaded = clusterName;
if (!ConfigConsts.NO_APPID_PLACEHOLDER.equalsIgnoreCase(appId)) {
//（核心邏輯，重點關注）加載配置信息
Release currentAppRelease = configService.loadConfig(appId, clientIp, appId, clusterName, namespace,
dataCenter, clientMessages);
if (currentAppRelease != null) {
releases.add(currentAppRelease);
//we have cluster search process, so the cluster name might be overridden
appClusterNameLoaded = currentAppRelease.getClusterName();
}
}
//if namespace does not belong to this appId, should check if there is a public configuration
// 如果namespace不屬于當前appId，而是屬于公共的配置文件。具體應用場景，就是應用A向共享自己的配置給其他應用，就可以將其自身的配置文件設置成public類型的
if (!namespaceBelongsToAppId(appId, namespace)) {
Release publicRelease = this.findPublicConfig(appId, clientIp, clusterName, namespace,
dataCenter, clientMessages);
if (Objects.nonNull(publicRelease)) {
releases.add(publicRelease);
}
}
if (releases.isEmpty()) {
response.sendError(HttpServletResponse.SC_NOT_FOUND,
String.format(
"Could not load configurations with appId: %s, clusterName: %s, namespace: %s",
appId, clusterName, originalNamespace));
Tracer.logEvent("Apollo.Config.NotFound",
assembleKey(appId, clusterName, originalNamespace, dataCenter));
return null;
}
auditReleases(appId, clusterName, dataCenter, clientIp, releases);
// 格式是：私有的ReleaseKey1+私有的ReleaseKey2+public的ReleaseKey1+public的ReleaseKey1
String mergedReleaseKey = releases.stream().map(Release::getReleaseKey)
.collect(Collectors.joining(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR));
// Client端上的ReleaseKey與Server端key相同，則沒有配置沒有變更
if (mergedReleaseKey.equals(clientSideReleaseKey)) {
// Client side configuration is the same with server side, return 304
response.setStatus(HttpServletResponse.SC_NOT_MODIFIED);
Tracer.logEvent("Apollo.Config.NotModified",
assembleKey(appId, appClusterNameLoaded, originalNamespace, dataCenter));
return null;
}
ApolloConfig apolloConfig = new ApolloConfig(appId, appClusterNameLoaded, originalNamespace,
mergedReleaseKey);
// 合并配置信息
apolloConfig.setConfigurations(mergeReleaseConfigurations(releases));
Tracer.logEvent("Apollo.Config.Found", assembleKey(appId, appClusterNameLoaded,
originalNamespace, dataCenter));
return apolloConfig;
}
}
3.2.3.2、AbstractConfigService#loadConfig
public abstract class AbstractConfigService implements ConfigService {
@Autowired
private GrayReleaseRulesHolder grayReleaseRulesHolder;
/**
* 加載配置
* @return
*/
@Override
public Release loadConfig(String clientAppId, String clientIp, String configAppId, String configClusterName,
String configNamespace, String dataCenter, ApolloNotificationMessages clientMessages) {
// load from specified cluster first
// 從指定cluster拉取配置
if (!Objects.equals(ConfigConsts.CLUSTER_NAME_DEFAULT, configClusterName)) {
// 查找配置
Release clusterRelease = findRelease(clientAppId, clientIp, configAppId, configClusterName, configNamespace,
clientMessages);
if (Objects.nonNull(clusterRelease)) {
return clusterRelease;
}
}
// try to load via data center
// 從指定的dataCenter的cluster加載配置
if (!Strings.isNullOrEmpty(dataCenter) && !Objects.equals(dataCenter, configClusterName)) {
Release dataCenterRelease = findRelease(clientAppId, clientIp, configAppId, dataCenter, configNamespace,
clientMessages);
if (Objects.nonNull(dataCenterRelease)) {
return dataCenterRelease;
}
}
// fallback to default release
// 不指定，走默認
return findRelease(clientAppId, clientIp, configAppId, ConfigConsts.CLUSTER_NAME_DEFAULT, configNamespace,
clientMessages);
}
/**
* 查找配置信息
*/
private Release findRelease(String clientAppId, String clientIp, String configAppId, String configClusterName,
String configNamespace, ApolloNotificationMessages clientMessages) {
// 獲取namespace的灰度發布的配置編號
Long grayReleaseId = grayReleaseRulesHolder.findReleaseIdFromGrayReleaseRule(clientAppId, clientIp, configAppId,
configClusterName, configNamespace);
Release release = null;
// 通過灰度的配置編號獲取具體配置信息
if (grayReleaseId != null) {
release = findActiveOne(grayReleaseId, clientMessages);
}
// 如果沒有灰度發布的信息，則直接獲取namespace最新的配置信息
if (release == null) {
//（核心邏輯，重點關注）通過appId + cluster + namespace，拉取最新配置信息
release = findLatestActiveRelease(configAppId, configClusterName, configNamespace, clientMessages);
}
return release;
}
}
3.2.3.3、ConfigServiceWithCache#findLatestActiveRelease
從Cache或DB中查詢namespace的最新配置信息
public class ConfigServiceWithCache extends AbstractConfigService {
...
private LoadingCache<String, ConfigCacheEntry> configCache;
private LoadingCache<Long, Optional<Release>> configIdCache;
@Override
protected Release findActiveOne(long id, ApolloNotificationMessages clientMessages) {
Tracer.logEvent(TRACER_EVENT_CACHE_GET_ID, String.valueOf(id));
return configIdCache.getUnchecked(id).orElse(null);
}
/**
* （核心邏輯，重點關注）拉取數據，本地緩存沒有就拉取DB
* @return
*/
@Override
protected Release findLatestActiveRelease(String appId, String clusterName, String namespaceName,
ApolloNotificationMessages clientMessages) {
String key = ReleaseMessageKeyGenerator.generate(appId, clusterName, namespaceName);
Tracer.logEvent(TRACER_EVENT_CACHE_GET, key);
// 獲取namespace對應的緩存配置信息，此處key為(appId+clusterName+namespaceName)，獲取到的信息包括通知編號(notificationId和具體配置信息Release)
ConfigCacheEntry cacheEntry = configCache.getUnchecked(key);
//cache is out-dated
// 緩存過期
if (clientMessages != null && clientMessages.has(key) &&
clientMessages.get(key) > cacheEntry.getNotificationId()) {
//invalidate the cache and try to load from db again
// 清除緩存（guava cache）
invalidate(key);
// 重新從DB中拉取緩存，獲取該namespace下的最新通知編號(notificationId)及其對應的配置信息
cacheEntry = configCache.getUnchecked(key);
}
// 如果緩存的版本信息大于當前客戶端所上傳的版本，則直接返回最新配置信息
return cacheEntry.getRelease();
}
private void invalidate(String key) {
configCache.invalidate(key);
Tracer.logEvent(TRACER_EVENT_CACHE_INVALIDATE, key);
}
@Override
public void handleMessage(ReleaseMessage message, String channel) {
logger.info("message received - channel: {}, message: {}", channel, message);
if (!Topics.APOLLO_RELEASE_TOPIC.equals(channel) || Strings.isNullOrEmpty(message.getMessage())) {
return;
}
try {
// 清除緩存
invalidate(message.getMessage());
//warm up the cache
// 重新從DB中拉取緩存，獲取該namespace下的最新通知編號(notificationId)及其對應的配置信息
configCache.getUnchecked(message.getMessage());
} catch (Throwable ex) {
//ignore
}
}
private static class ConfigCacheEntry {
private final long notificationId;
private final Release release;
public ConfigCacheEntry(long notificationId, Release release) {
this.notificationId = notificationId;
this.release = release;
}
public long getNotificationId() {
return notificationId;
}
public Release getRelease() {
return release;
}
}
}
四、最后
《碼頭工人的一千零一夜》是一位專注于技術干貨分享的博主，追隨博主的文章，你將深入了解業界最新的技術趨勢，以及在Java開發和安全領域的實用經驗分享。無論你是開發人員還是對逆向工程感興趣的愛好者，都能在《碼頭工人的一千零一夜》找到有價值的知識和見解。
懂得不多，做得太少。歡迎批評、指正。
浙公網安備 33010602011771號