ApachePulsar中TopicLookup请求处理的逻辑是什么
本篇内容主要讲解“Apache Pulsar中TopicLookup请求处理的逻辑是什么”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Apache Pulsar中TopicLookup请求处理的逻辑是什么”吧!
创新互联2013年开创至今,是专业互联网技术服务公司,拥有项目成都网站设计、成都网站制作、外贸网站建设网站策划,项目实施与项目整合能力。我们以让每一个梦想脱颖而出为使命,1280元梨树做网站,已为上家服务,为梨树各地企业和个人服务,联系电话:13518219792
实际的核心逻辑是这2行代码
LookupOptions options = LookupOptions.builder() .authoritative(authoritative) .advertisedListenerName(advertisedListenerName) .loadTopicsInBundle(true) // 这里这个条件是true .build(); pulsarService.getNamespaceService().getBrokerServiceUrlAsync(topicName, options)
这里传递的参数将loadTopicsInBundle
设置了成true。我们看下在处理lookup请求过程中是否有loadtopic的逻辑。
NamespaceService.findBrokerServiceUrl
这个函数我们注意到有 ownershipCache.getOwnerAsync
和searchForCandidateBroker
这2个地方没有细说
我们先看一下ownershipCache
。
private CompletableFuture> findBrokerServiceUrl( NamespaceBundle bundle, LookupOptions options) { .... return targetMap.computeIfAbsent(bundle, (k) -> { ... ownershipCache.getOwnerAsync(bundle) .thenAccept(nsData -> { // nsData : Optional if (!nsData.isPresent()) { ... // 目前还没有人负责这个bundle 尝试查找这个bundle的owner pulsar.getExecutor().execute(() -> { searchForCandidateBroker(bundle, future, options); }); ... } ... }); }
OwnerShipCache类
从javadoc 里面可以知道这个类的主要功能。
cache zk里面关于 service unit 的ownership信息
提供zk的读写功能
可以用来查找owner信息
可以用来获取一个 service unit 的ownership
getOwnerAsync 这个方法主要是查看zk cache里面是否有信息,如果没有信息,则尝试读取zk节点,
如果节点有信息则说明有人拿到了这个bundle的ownership
如果这个节点就是当前机器,则会通知bundle load的信息给listener
如果这个节点没有信息,说明当前还没有人负责这个bundle。
// org.apache.pulsar.broker.namespace.OwnerShipCache public CompletableFuture> getOwnerAsync(NamespaceBundle suName) { // 这里的路径是 /namespace/{namespace}/0x{lowerEndpoint}_0x{upperEndpoint} String path = ServiceUnitZkUtils.path(suName); // ownedBundleFuture 还是一个 AsyncLoadingCache // 这里不会尝试去加载这个cache信息,因为调用的getIfPresent CompletableFuture ownedBundleFuture = ownedBundlesCache.getIfPresent(path); // 如果之前有内容的话就说明当前broker是owner(这部分逻辑在cache的加载代码里面,后面会说) 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里面没有,我们确认下当前的owner是谁。 // 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) // 这个逻辑是从zk里面读取这个bundle路径下的内容 .thenApply(optionalOwnerDataWithStat -> { // 如果这个路径下有数据,则说明有人已经成功获取了这个bundle的ownership信息 if (optionalOwnerDataWithStat.isPresent()) { Map.Entry ownerDataWithStat = optionalOwnerDataWithStat.get(); Stat stat = ownerDataWithStat.getValue(); // 如果这个zk临时节点的owner就是当前的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); // 这里会通知callback(和主要逻辑无关) namespaceService.onNamespaceBundleOwned(ownedBundle.getNamespaceBundle()); } } // 这里返回的是一个Optional对象,如果这个节点不存在的话返回的实际是一个Empty // 说明这个时候没有人负责这个bundle // 也可能返回带有信息的optional,这时候负责这个节点的broker可能是当前机器也可能是其他机器。 return optionalOwnerDataWithStat; }); }
我们看一下如果没有任何人负责这个bundle的情况。
NamespaceService.searchForCandidateBroker
这个方法的逻辑是选出当前这个bundle的owner是哪个broker
主要依靠LeaderElectionService
和LoadManager
选出。
如果选出来的broker是本机的话,则会尝试获取这个bundle的ownership。
如果是其他机器的话则会把这个请求转发给其他机器,请求其他机器来获取ownership。
private void searchForCandidateBroker(NamespaceBundle bundle, CompletableFuture> lookupFuture, LookupOptions options) { ... // 首先会按照一定逻辑来选出这个bundle的可能的broker节点 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 // 如果这个LeaderElectionService 是leader || // 不是中心化的loadManager(这个是均衡负载用的)|| // 如果当前这个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选一个负载最轻的broker出来 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) { ... } // 到这里就选出一个候选的broker地址了 try { checkNotNull(candidateBroker); // 如果这个候选broker就是当前机器 if (candidateBroker.equals(pulsar.getSafeWebServiceAddress())) { ... // 这里使用ownerShipCache尝试获取这个bundle的ownership ownershipCache.tryAcquiringOwnership(bundle) .thenAccept(ownerInfo -> { ... // 这里就是文章开始的时候说的是否需要load 所有在bundle里面的topic if (options.isLoadTopicsInBundle()) { // Schedule the task to pre-load topics pulsar.loadNamespaceTopics(bundle); } // find the target // 走到这里说明已经把当前的broker作为这个bundle的owner了,直接返回本机的信息给请求者 lookupFuture.complete(Optional.of(new LookupResult(ownerInfo))); return; } }).exceptionally(exception -> { ... }); } else { ... // 这里是把这个lookup 请求转发给其他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
这里就是尝试获取这个bundle的ownership的逻辑了。
只需要在zk上记录当前节点的信息就可以了。
(也会有维护这个cache的逻辑)
public CompletableFuturetryAcquiringOwnership(NamespaceBundle bundle) throws Exception { String path = ServiceUnitZkUtils.path(bundle); CompletableFuture future = new CompletableFuture<>(); ... LOG.info("Trying to acquire ownership of {}", bundle); // 这里调用的是get,这个方法会触发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 -> { // 到这里说明已经获得了这个bundle的ownership了,直接返回。 LOG.info("Successfully acquired ownership of {}", path); namespaceService.onNamespaceBundleOwned(bundle); future.complete(selfOwnerInfo); }).exceptionally(exception -> { // 这里如果加载过程中出现问题(可能是其他人成为了leader) // Failed to acquire ownership if (exception instanceof CompletionException && exception.getCause() instanceof KeeperException.NodeExistsException) { // 确认当前的leader是谁 resolveOwnership(path) .thenAccept(optionalOwnerDataWithStat -> { // 这里会拿到之前成功获得ownership的节点信息 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 加载逻辑
这里逻辑比较简单,序列化本机的连接信息,写入到这个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
到这里我们已经可以拿到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)) { // 到这里会创建一个Topic对象保存在BrokerService里面 // 这部分后面会说,涉及到 ManagedLedger 里面的初始化 CompletableFuture future = brokerService.getOrCreateTopic(topic); if (future != null) { persistentTopics.add(future); } } } ... } ... return null; }); }
NamespaceService.getListOfPersistentTopics
这里就比较容易了
读取zk的/managed-ledgers/%s/persistent
所有子节点即可。
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; }); }
到此,相信大家对“Apache Pulsar中TopicLookup请求处理的逻辑是什么”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
本文标题:ApachePulsar中TopicLookup请求处理的逻辑是什么
文章路径:http://myzitong.com/article/jddcsc.html