Apache Pulsar中TopicLookup请求处理的逻辑是什么

本篇内容主要讲解"Apache Pulsar中TopicLookup请求处理的逻辑是什么"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"Apache Pulsar中TopicLookup请求处理的逻辑是什么"吧!

实际的核心逻辑是这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.OwnerShipCachepublic 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 CompletableFuture   tryAcquiringOwnership(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请求处理的逻辑是什么"有了更深的了解，不妨来实际操作一番吧！这里是网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！