溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
本篇內(nèi)容主要講解“Apache Pulsar中TopicLookup請(qǐng)求處理的邏輯是什么”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Apache Pulsar中TopicLookup請(qǐng)求處理的邏輯是什么”吧!
實(shí)際的核心邏輯是這2行代碼
LookupOptions options = LookupOptions.builder() .authoritative(authoritative) .advertisedListenerName(advertisedListenerName) .loadTopicsInBundle(true) // 這里這個(gè)條件是true .build(); pulsarService.getNamespaceService().getBrokerServiceUrlAsync(topicName, options)
這里傳遞的參數(shù)將loadTopicsInBundle 設(shè)置了成true。我們看下在處理lookup請(qǐng)求過(guò)程中是否有l(wèi)oadtopic的邏輯。
這個(gè)函數(shù)我們注意到有 ownershipCache.getOwnerAsync 和searchForCandidateBroker 這2個(gè)地方?jīng)]有細(xì)說(shuō)
我們先看一下ownershipCache。
private CompletableFuture<Optional<LookupResult>> findBrokerServiceUrl(
NamespaceBundle bundle, LookupOptions options) {
....
return targetMap.computeIfAbsent(bundle, (k) -> {
...
ownershipCache.getOwnerAsync(bundle)
.thenAccept(nsData -> {
// nsData : Optional<NamespaceEphemeralData>
if (!nsData.isPresent()) {
...
// 目前還沒有人負(fù)責(zé)這個(gè)bundle 嘗試查找這個(gè)bundle的owner
pulsar.getExecutor().execute(() -> {
searchForCandidateBroker(bundle, future, options);
});
...
}
...
});
}從javadoc 里面可以知道這個(gè)類的主要功能。
cache zk里面關(guān)于 service unit 的ownership信息
提供zk的讀寫功能
可以用來(lái)查找owner信息
可以用來(lái)獲取一個(gè) service unit 的ownership
getOwnerAsync 這個(gè)方法主要是查看zk cache里面是否有信息，如果沒有信息，則嘗試讀取zk節(jié)點(diǎn)，
如果節(jié)點(diǎn)有信息則說(shuō)明有人拿到了這個(gè)bundle的ownership
如果這個(gè)節(jié)點(diǎn)就是當(dāng)前機(jī)器，則會(huì)通知bundle load的信息給listener
如果這個(gè)節(jié)點(diǎn)沒有信息，說(shuō)明當(dāng)前還沒有人負(fù)責(zé)這個(gè)bundle。
// org.apache.pulsar.broker.namespace.OwnerShipCache
public
CompletableFuture<Optional<NamespaceEphemeralData>>
getOwnerAsync(NamespaceBundle suName)
{
// 這里的路徑是 /namespace/{namespace}/0x{lowerEndpoint}_0x{upperEndpoint}
String path = ServiceUnitZkUtils.path(suName);
// ownedBundleFuture 還是一個(gè) AsyncLoadingCache
// 這里不會(huì)嘗試去加載這個(gè)cache信息，因?yàn)檎{(diào)用的getIfPresent
CompletableFuture<OwnedBundle> ownedBundleFuture = ownedBundlesCache.getIfPresent(path);
// 如果之前有內(nèi)容的話就說(shuō)明當(dāng)前broker是owner（這部分邏輯在cache的加載代碼里面，后面會(huì)說(shuō)）
if (ownedBundleFuture != null) {
// Either we're the owners or we're trying to become the owner.
return ownedBundleFuture.thenApply(serviceUnit -> {
// We are the owner of the service unit
return Optional.of(serviceUnit.isActive() ? selfOwnerInfo : selfOwnerInfoDisabled);
});
}
// 如果cache里面沒有，我們確認(rèn)下當(dāng)前的owner是誰(shuí)。
// If we're not the owner, we need to check if anybody else is
return resolveOwnership(path)
.thenApply(optional -> optional.map(Map.Entry::getKey));
}
private CompletableFuture<Optional<Map.Entry<NamespaceEphemeralData, Stat>>> resolveOwnership(String path) {
return ownershipReadOnlyCache.getWithStatAsync(path) // 這個(gè)邏輯是從zk里面讀取這個(gè)bundle路徑下的內(nèi)容
.thenApply(optionalOwnerDataWithStat -> {
// 如果這個(gè)路徑下有數(shù)據(jù)，則說(shuō)明有人已經(jīng)成功獲取了這個(gè)bundle的ownership信息
if (optionalOwnerDataWithStat.isPresent()) {
Map.Entry<NamespaceEphemeralData, Stat> ownerDataWithStat = optionalOwnerDataWithStat.get();
Stat stat = ownerDataWithStat.getValue();
// 如果這個(gè)zk臨時(shí)節(jié)點(diǎn)的owner就是當(dāng)前的broker
if (stat.getEphemeralOwner() == localZkCache.getZooKeeper().getSessionId()) {
LOG.info("Successfully reestablish ownership of {}", path);
// 這里是更新緩存的邏輯
OwnedBundle ownedBundle = new OwnedBundle(ServiceUnitZkUtils.suBundleFromPath(path, bundleFactory));
if (selfOwnerInfo.getNativeUrl().equals(ownerDataWithStat.getKey().getNativeUrl())) {
ownedBundlesCache.put(path, CompletableFuture.completedFuture(ownedBundle));
}
ownershipReadOnlyCache.invalidate(path);
// 這里會(huì)通知callback（和主要邏輯無(wú)關(guān)）
namespaceService.onNamespaceBundleOwned(ownedBundle.getNamespaceBundle());
}
}
// 這里返回的是一個(gè)Optional對(duì)象，如果這個(gè)節(jié)點(diǎn)不存在的話返回的實(shí)際是一個(gè)Empty
// 說(shuō)明這個(gè)時(shí)候沒有人負(fù)責(zé)這個(gè)bundle
// 也可能返回帶有信息的optional，這時(shí)候負(fù)責(zé)這個(gè)節(jié)點(diǎn)的broker可能是當(dāng)前機(jī)器也可能是其他機(jī)器。
return optionalOwnerDataWithStat;
});
}我們看一下如果沒有任何人負(fù)責(zé)這個(gè)bundle的情況。
這個(gè)方法的邏輯是選出當(dāng)前這個(gè)bundle的owner是哪個(gè)broker
主要依靠LeaderElectionService 和LoadManager 選出。
如果選出來(lái)的broker是本機(jī)的話，則會(huì)嘗試獲取這個(gè)bundle的ownership。
如果是其他機(jī)器的話則會(huì)把這個(gè)請(qǐng)求轉(zhuǎn)發(fā)給其他機(jī)器，請(qǐng)求其他機(jī)器來(lái)獲取ownership。
private void searchForCandidateBroker(NamespaceBundle bundle,
CompletableFuture<Optional<LookupResult>> lookupFuture,
LookupOptions options) {
...
// 首先會(huì)按照一定邏輯來(lái)選出這個(gè)bundle的可能的broker節(jié)點(diǎn)
String candidateBroker = null;
...
boolean authoritativeRedirect = les.isLeader();
try {
// check if this is Heartbeat or SLAMonitor namespace
...
if (candidateBroker == null) {
if (options.isAuthoritative()) {
// leader broker already assigned the current broker as owner
candidateBroker = pulsar.getSafeWebServiceAddress();
} else
// 如果這個(gè)LeaderElectionService 是leader ||
// 不是中心化的loadManager（這個(gè)是均衡負(fù)載用的）||
// 如果當(dāng)前這個(gè)leader的broker還不是active的
if (!this.loadManager.get().isCentralized()
|| pulsar.getLeaderElectionService().isLeader()
// If leader is not active, fallback to pick the least loaded from current broker loadmanager
|| !isBrokerActive(pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl())
) {
// 從loadManager選一個(gè)負(fù)載最輕的broker出來(lái)
Optional<String> availableBroker = getLeastLoadedFromLoadManager(bundle);
if (!availableBroker.isPresent()) {
lookupFuture.complete(Optional.empty());
return;
}
candidateBroker = availableBroker.get();
authoritativeRedirect = true;
} else {
// forward to leader broker to make assignment
candidateBroker = pulsar.getLeaderElectionService().getCurrentLeader().getServiceUrl();
}
}
} catch (Exception e) {
...
}
// 到這里就選出一個(gè)候選的broker地址了
try {
checkNotNull(candidateBroker);
// 如果這個(gè)候選broker就是當(dāng)前機(jī)器
if (candidateBroker.equals(pulsar.getSafeWebServiceAddress())) {
...
// 這里使用ownerShipCache嘗試獲取這個(gè)bundle的ownership
ownershipCache.tryAcquiringOwnership(bundle)
.thenAccept(ownerInfo -> {
...
// 這里就是文章開始的時(shí)候說(shuō)的是否需要load 所有在bundle里面的topic
if (options.isLoadTopicsInBundle()) {
// Schedule the task to pre-load topics
pulsar.loadNamespaceTopics(bundle);
}
// find the target
// 走到這里說(shuō)明已經(jīng)把當(dāng)前的broker作為這個(gè)bundle的owner了，直接返回本機(jī)的信息給請(qǐng)求者
lookupFuture.complete(Optional.of(new LookupResult(ownerInfo)));
return;
}
}).exceptionally(exception -> {
...
});
} else {
...
// 這里是把這個(gè)lookup 請(qǐng)求轉(zhuǎn)發(fā)給其他broker
// Load managed decider some other broker should try to acquire ownership
// Now setting the redirect url
createLookupResult(candidateBroker, authoritativeRedirect, options.getAdvertisedListenerName())
.thenAccept(lookupResult -> lookupFuture.complete(Optional.of(lookupResult)))
.exceptionally(ex -> {
lookupFuture.completeExceptionally(ex);
return null;
});
}
} catch (Exception e) {
...
}
}這里就是嘗試獲取這個(gè)bundle的ownership的邏輯了。
只需要在zk上記錄當(dāng)前節(jié)點(diǎn)的信息就可以了。
（也會(huì)有維護(hù)這個(gè)cache的邏輯）
public CompletableFuture<NamespaceEphemeralData>
tryAcquiringOwnership(NamespaceBundle bundle) throws Exception {
String path = ServiceUnitZkUtils.path(bundle);
CompletableFuture<NamespaceEphemeralData> future = new CompletableFuture<>();
...
LOG.info("Trying to acquire ownership of {}", bundle);
// 這里調(diào)用的是get，這個(gè)方法會(huì)觸發(fā)cache加載的邏輯。
// Doing a get() on the ownedBundlesCache will trigger an async ZK write to acquire the lock over the
// service unit
ownedBundlesCache.get(path)
.thenAccept(namespaceBundle -> {
// 到這里說(shuō)明已經(jīng)獲得了這個(gè)bundle的ownership了,直接返回。
LOG.info("Successfully acquired ownership of {}", path);
namespaceService.onNamespaceBundleOwned(bundle);
future.complete(selfOwnerInfo);
}).exceptionally(exception -> {
// 這里如果加載過(guò)程中出現(xiàn)問題（可能是其他人成為了leader）
// Failed to acquire ownership
if (exception instanceof CompletionException
&& exception.getCause() instanceof KeeperException.NodeExistsException) {
// 確認(rèn)當(dāng)前的leader是誰(shuí)
resolveOwnership(path)
.thenAccept(optionalOwnerDataWithStat -> {
// 這里會(huì)拿到之前成功獲得ownership的節(jié)點(diǎn)信息
if (optionalOwnerDataWithStat.isPresent()) {
Map.Entry<NamespaceEphemeralData, Stat> ownerDataWithStat = optionalOwnerDataWithStat.get();
NamespaceEphemeralData ownerData = ownerDataWithStat.getKey();
Stat stat = ownerDataWithStat.getValue();
if (stat.getEphemeralOwner() != localZkCache.getZooKeeper().getSessionId()) {
LOG.info("Failed to acquire ownership of {} -- Already owned by broker {}",
path, ownerData);
}
// 直接返回即可
future.complete(ownerData);
} else {
...
}{
}).exceptionally(ex -> {
....
});
} else {
...
}
return null;
});
return future;
}這里邏輯比較簡(jiǎn)單，序列化本機(jī)的連接信息，寫入到這個(gè)bundle的path下面就行了
private class OwnedServiceUnitCacheLoader implements AsyncCacheLoader<String, OwnedBundle> {
@SuppressWarnings("deprecation")
@Override
public CompletableFuture<OwnedBundle> asyncLoad(String namespaceBundleZNode, Executor executor) {
if (LOG.isDebugEnabled()) {
LOG.debug("Acquiring zk lock on namespace {}", namespaceBundleZNode);
}
byte[] znodeContent;
try {
znodeContent = jsonMapper.writeValueAsBytes(selfOwnerInfo);
} catch (JsonProcessingException e) {
// Failed to serialize to JSON
return FutureUtil.failedFuture(e);
}
CompletableFuture<OwnedBundle> future = new CompletableFuture<>();
ZkUtils.asyncCreateFullPathOptimistic(localZkCache.getZooKeeper(), namespaceBundleZNode, znodeContent,
Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL, (rc, path, ctx, name) -> {
if (rc == KeeperException.Code.OK.intValue()) {
if (LOG.isDebugEnabled()) {
LOG.debug("Successfully acquired zk lock on {}", namespaceBundleZNode);
}
ownershipReadOnlyCache.invalidate(namespaceBundleZNode);
future.complete(new OwnedBundle(
ServiceUnitZkUtils.suBundleFromPath(namespaceBundleZNode, bundleFactory)));
} else {
// Failed to acquire lock
future.completeExceptionally(KeeperException.create(rc));
}
}, null);
return future;
}
}到這里我們已經(jīng)可以拿到bundle的ownership了。我們看一下之前加載所有topic的邏輯。
PulsarService.loadNamespaceTopics
public void loadNamespaceTopics(NamespaceBundle bundle) {
executor.submit(() -> {
NamespaceName nsName = bundle.getNamespaceObject();
List<CompletableFuture<Topic>> persistentTopics = Lists.newArrayList();
long topicLoadStart = System.nanoTime();
for (String topic : getNamespaceService().getListOfPersistentTopics(nsName).join()) {
try {
TopicName topicName = TopicName.get(topic);
if (bundle.includes(topicName)) {
// 到這里會(huì)創(chuàng)建一個(gè)Topic對(duì)象保存在BrokerService里面
// 這部分后面會(huì)說(shuō)，涉及到 ManagedLedger 里面的初始化
CompletableFuture<Topic> future = brokerService.getOrCreateTopic(topic);
if (future != null) {
persistentTopics.add(future);
}
}
}
...
}
...
return null;
});
}NamespaceService.getListOfPersistentTopics
這里就比較容易了
讀取zk的/managed-ledgers/%s/persistent所有子節(jié)點(diǎn)即可。
public CompletableFuture<List<String>> getListOfPersistentTopics(NamespaceName namespaceName) {
// For every topic there will be a managed ledger created.
String path = String.format("/managed-ledgers/%s/persistent", namespaceName);
if (LOG.isDebugEnabled()) {
LOG.debug("Getting children from managed-ledgers now: {}", path);
}
return pulsar.getLocalZkCacheService().managedLedgerListCache().getAsync(path)
.thenApply(znodes -> {
List<String> topics = Lists.newArrayList();
for (String znode : znodes) {
topics.add(String.format("persistent://%s/%s", namespaceName, Codec.decode(znode)));
}
topics.sort(null);
return topics;
});
}到此，相信大家對(duì)“Apache Pulsar中TopicLookup請(qǐng)求處理的邏輯是什么”有了更深的了解，不妨來(lái)實(shí)際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼