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