本篇內(nèi)容主要講解“Apache Pulsar中TopicLookup請(qǐng)求處理的邏輯是什么”，感興趣的朋友不妨來(lái)看看。本文介紹的方法操作簡(jiǎn)單快捷，實(shí)用性強(qiáng)。下面就讓小編來(lái)帶大家學(xué)習(xí)“Apache Pulsar中TopicLookup請(qǐng)求處理的邏輯是什么”吧!
創(chuàng)新互聯(lián)2013年開(kāi)創(chuàng)至今，是專業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司，擁有項(xiàng)目成都網(wǎng)站設(shè)計(jì)、成都網(wǎng)站制作、外貿(mào)網(wǎng)站建設(shè)網(wǎng)站策劃，項(xiàng)目實(shí)施與項(xiàng)目整合能力。我們以讓每一個(gè)夢(mèng)想脫穎而出為使命，1280元梨樹(shù)做網(wǎng)站,已為上家服務(wù),為梨樹(shù)各地企業(yè)和個(gè)人服務(wù),聯(lián)系電話:13518219792
實(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的邏輯。
NamespaceService.findBrokerServiceUrl
這個(gè)函數(shù)我們注意到有 ownershipCache.getOwnerAsync 和searchForCandidateBroker 這2個(gè)地方?jīng)]有細(xì)說(shuō)
我們先看一下ownershipCache。
private CompletableFuture> findBrokerServiceUrl(
NamespaceBundle bundle, LookupOptions options) {
....
return targetMap.computeIfAbsent(bundle, (k) -> {
...
ownershipCache.getOwnerAsync(bundle)
.thenAccept(nsData -> {
// nsData : Optional
if (!nsData.isPresent()) {
...
// 目前還沒(méi)有人負(fù)責(zé)這個(gè)bundle 嘗試查找這個(gè)bundle的owner
pulsar.getExecutor().execute(() -> {
searchForCandidateBroker(bundle, future, options);
});
...
}
...
});
}
OwnerShipCache類
從javadoc 里面可以知道這個(gè)類的主要功能。
getOwnerAsync 這個(gè)方法主要是查看zk cache里面是否有信息，如果沒(méi)有信息，則嘗試讀取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)沒(méi)有信息，說(shuō)明當(dāng)前還沒(méi)有人負(fù)責(zé)這個(gè)bundle。
// org.apache.pulsar.broker.namespace.OwnerShipCache
public
CompletableFuture>
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 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里面沒(méi)有，我們確認(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>> 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 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í)候沒(méi)有人負(fù)責(zé)這個(gè)bundle
// 也可能返回帶有信息的optional，這時(shí)候負(fù)責(zé)這個(gè)節(jié)點(diǎn)的broker可能是當(dāng)前機(jī)器也可能是其他機(jī)器。
return optionalOwnerDataWithStat;
});
}
我們看一下如果沒(méi)有任何人負(fù)責(zé)這個(gè)bundle的情況。
NamespaceService.searchForCandidateBroker
這個(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> 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 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 -> {
...
// 這里就是文章開(kāi)始的時(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) {
...
}
}
OwnershipCache.tryAcquiringOwnership
這里就是嘗試獲取這個(gè)bundle的ownership的邏輯了。
只需要在zk上記錄當(dāng)前節(jié)點(diǎn)的信息就可以了。
（也會(huì)有維護(hù)這個(gè)cache的邏輯）
public CompletableFuture
tryAcquiringOwnership(NamespaceBundle bundle) throws Exception {
String path = ServiceUnitZkUtils.path(bundle);
CompletableFuture 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)問(wè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 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;
}
OwnershipCache 加載邏輯
這里邏輯比較簡(jiǎn)單，序列化本機(jī)的連接信息，寫(xiě)入到這個(gè)bundle的path下面就行了
private class OwnedServiceUnitCacheLoader implements AsyncCacheLoader {
@SuppressWarnings("deprecation")
@Override
public CompletableFuture 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 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;
}
}
加載bundle下所有topic
到這里我們已經(jīng)可以拿到bundle的ownership了。我們看一下之前加載所有topic的邏輯。
PulsarService.loadNamespaceTopics
public void loadNamespaceTopics(NamespaceBundle bundle) {
executor.submit(() -> {
NamespaceName nsName = bundle.getNamespaceObject();
List> 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 future = brokerService.getOrCreateTopic(topic);
if (future != null) {
persistentTopics.add(future);
}
}
}
...
}
...
return null;
});
}NamespaceService.getListOfPersistentTopics
這里就比較容易了
讀取zk的/managed-ledgers/%s/persistent所有子節(jié)點(diǎn)即可。
public CompletableFuture> 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 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í)際操作一番吧！這里是創(chuàng)新互聯(lián)網(wǎng)站，更多相關(guān)內(nèi)容可以進(jìn)入相關(guān)頻道進(jìn)行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！
文章標(biāo)題：ApachePulsar中TopicLookup請(qǐng)求處理的邏輯是什么
文章網(wǎng)址：
http://weahome.cn/article/jddcsc.html
重慶分公司
重慶分公司
