Apache Pulsar源码解析之Lookup机制

news/2024/7/7 19:50:44

引言

在学习Pulsar一段时间后，相信大家也或多或少听说Lookup这个词，今天就一起来深入剖析下Pulsar是怎么设计的它吧

Lookup是什么

在客户端跟服务端建立TCP连接前有些信息需要提前获取，这个获取方式就是Lookup机制。所获取的信息有以下几种

  • 应该跟哪台Broker建立连接
  • Topic的Schema信息
  • Topic的分区信息

其中第一个是最重要的，因此今天就针对第一点进行深入剖析，大致流程如下图
在这里插入图片描述

  1. 在创建生产者/消费者时会触发Lookup，一般是通过HTTP请求Broker来获取目标Topic所归属的Broker节点信息，这样才知道跟哪台机器建立TCP连接进行数据交互
  2. Broker接收到Lookup命令，此时会进行限流检查、身份/权限认证、校验集群等检测动作后，根据请求中携带的Namespace信息获取对应的Namespace对象进行处理，这里Namespace会对Topic进行哈希运算并判断它落在数组的哪一个节点，算出来后就根据数组的信息来从Bundle数组中获得对应的Bundle，这个过程其实就是一致性哈希算法寻址过程。
  3. 在获得Bundle后会尝试从本机Cache中查询该Bundle所归属的Broker信息。
  4. 如果在Cache中没有命中，则会去Zookeeper中进行读取，如果发现该Bundle还未归属Broker则触发归属Broker的流程
  5. 获取到该Topic所归属的Broker信息后返回给客户端，客户端解析结果并跟所归属的Broker建立TCP连接，用于后续生产者往Broker节点进行消息写入

补充说明确定Bundle的归属，如果Broker的loadManager使用的是中心化策略，则需要Broker Leader来当裁判决定，否则当前Broker就可当作裁判。虽然Broker是无状态的，但会通过Zookeeper选举出一个Leader用于监控负载、为Bundle分配Broker等事情，裁判Broker通过loadManager查找负载最低的Broker并把Bundle分配给它。

客户端实现原理

Lookup机制是由客户端发起的，在创建生产者/消费者对象时会初始化网络连接，以生产者代码为例进行跟踪看看。无论是创建分区还是非分区生产者，最终都会走到ProducerImpl的构造函数，就从这里开始看吧

   public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
                        CompletableFuture<Producer<T>> producerCreatedFuture, int partitionIndex, Schema<T> schema,
                        ProducerInterceptors interceptors, Optional<String> overrideProducerName) {
				....
        //这里进去就是创建跟Broker的网络连接
        grabCnx();
    }

    void grabCnx() {
      	//实际上是调用ConnectionHandler进行的
        this.connectionHandler.grabCnx();
    }

		protected void grabCnx(Optional<URI> hostURI) {
  		....
      //这里是核心，相当于最终又调用回PulsarClientImpl类的getConnection方法
      cnxFuture = state.client.getConnection(state.topic, (state.redirectedClusterURI.toString()));
  		....
    }


    public CompletableFuture<ClientCnx> getConnection(final String topic, final String url) {
        TopicName topicName = TopicName.get(topic);
      	//看到方法名就知道到了Lookup的时候了，所以说好的命名远胜于注释
        return getLookup(url).getBroker(topicName)
                .thenCompose(lookupResult -> getConnection(lookupResult.getLogicalAddress(),
                        lookupResult.getPhysicalAddress(), cnxPool.genRandomKeyToSelectCon()));
    }

    public LookupService getLookup(String serviceUrl) {
        return urlLookupMap.computeIfAbsent(serviceUrl, url -> {
            try {
              	//忽略其他的，直接跟这里进去
                return createLookup(serviceUrl);
            } catch (PulsarClientException e) {
                log.warn("Failed to update url to lookup service {}, {}", url, e.getMessage());
                throw new IllegalStateException("Failed to update url " + url);
            }
        });
    }

    public LookupService createLookup(String url) throws PulsarClientException {
      	//这里可以看到如果咱们在配置客户端的地址是http开头就会通过http方式进行Loopup，否则走二进制协议进行查询
        if (url.startsWith("http")) {
            return new HttpLookupService(conf, eventLoopGroup);
        } else {
            return new BinaryProtoLookupService(this, url, conf.getListenerName(), conf.isUseTls(),
                    externalExecutorProvider.getExecutor());
        }
    }

		
    public HttpLookupService(ClientConfigurationData conf, EventLoopGroup eventLoopGroup)
            throws PulsarClientException {
      	//进到可能会误会Pulsar是通过HttpClient工具包进行的HTTP通信，继续看HttpClient构造函数
        this.httpClient = new HttpClient(conf, eventLoopGroup);
        this.useTls = conf.isUseTls();
        this.listenerName = conf.getListenerName();
    }

    protected HttpClient(ClientConfigurationData conf, EventLoopGroup eventLoopGroup) throws PulsarClientException {
			....
      //可以看到实际上最终是调用的AsyncHttpClient进行HTTP通信，这是一个封装Netty的async-http-client-2.12.1.jar的外部包
      httpClient = new DefaultAsyncHttpClient(config);
			....
    }

通过上面可以看到Lookup服务已经完成初始化，接下来就来看看客户端如何发起Lookup请求，回到PulsarClientImpl的getConnection方法，可以看到这里是链式调用，上面是从getLookup看到了其实是对Lookup进行初始化的过程，那么接下来就跟踪getBroker方法看看是怎么获取的服务端信息

    public CompletableFuture<ClientCnx> getConnection(final String topic, final String url) {
        TopicName topicName = TopicName.get(topic);
        return getLookup(url).getBroker(topicName)
                .thenCompose(lookupResult -> getConnection(lookupResult.getLogicalAddress(),
                        lookupResult.getPhysicalAddress(), cnxPool.genRandomKeyToSelectCon()));
    }

   public CompletableFuture<LookupTopicResult> getBroker(TopicName topicName) {
        //判断访问哪个版本的接口
        String basePath = topicName.isV2() ? BasePathV2 : BasePathV1;
        String path = basePath + topicName.getLookupName();
        path = StringUtils.isBlank(listenerName) ? path : path + "?listenerName=" + Codec.encode(listenerName);
        //获取要访问的Broker地址
        return httpClient.get(path, LookupData.class)
                .thenCompose(lookupData -> {
            URI uri = null;
            try {
              	//解析服务端返回的数据，本质上就是返回的就是Topic所在Broker的节点IP+端口
                InetSocketAddress brokerAddress = InetSocketAddress.createUnresolved(uri.getHost(), uri.getPort());
                //HTTP通过Lookup方式访问服务端绝对不会走代理
                return CompletableFuture.completedFuture(new LookupTopicResult(brokerAddress, brokerAddress,
                        false /* HTTP lookups never use the proxy */));
            } catch (Exception e) {
							....
            }
        });
    }

public class LookupTopicResult {
  	//LookupTopicResult是查询Topic归属Broker的结果后包装的一层结果，可以看到这里其实就是Socket信息也就是IP+端口
    private final InetSocketAddress logicalAddress;
    private final InetSocketAddress physicalAddress;
    private final boolean isUseProxy;
}

客户端的流程走到这里基本就结束了，是否有些意犹未尽迫不及待的想知道服务端又是怎么处理的？那么就看看下一节

服务端实现原理

服务端的入口在TopicLookup类的lookupTopicAsync方法，服务端大致步骤是这样的：1. 获取Topic所归属的Bundle 2. 查询Bundle所归属的Broker 3. 返回该Broker的url

    public void lookupTopicAsync(
            @Suspended AsyncResponse asyncResponse,
            @PathParam("topic-domain") String topicDomain, @PathParam("tenant") String tenant,
            @PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
            @QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
            @QueryParam("listenerName") String listenerName,
            @HeaderParam(LISTENERNAME_HEADER) String listenerNameHeader) {
        TopicName topicName = getTopicName(topicDomain, tenant, namespace, encodedTopic);
        if (StringUtils.isEmpty(listenerName) && StringUtils.isNotEmpty(listenerNameHeader)) {
            listenerName = listenerNameHeader;
        }
        //可以看得到这里是获取Lookup的，跟踪进去看看
        internalLookupTopicAsync(topicName, authoritative, listenerName)
                .thenAccept(lookupData -> asyncResponse.resume(lookupData))
                .exceptionally(ex -> {
                    ....
                });
    }


 			protected CompletableFuture<LookupData> internalLookupTopicAsync(final TopicName topicName, boolean authoritative, String listenerName) {
CompletableFuture<Optional<LookupResult>> lookupFuture = pulsar().getNamespaceService()
                            //获得目标Broker地址, 继续从这里进去
                            .getBrokerServiceUrlAsync(topicName,
                                    LookupOptions.builder()
                                            .advertisedListenerName(listenerName)
                                            .authoritative(authoritative)
                                            .loadTopicsInBundle(false)
                                            .build());
    }

    public CompletableFuture<Optional<LookupResult>> getBrokerServiceUrlAsync(TopicName topic, LookupOptions options) {
        long startTime = System.nanoTime();

        // 获取这个Topic所归属的Bundle
        CompletableFuture<Optional<LookupResult>> future = getBundleAsync(topic)
                .thenCompose(bundle -> {
                    //根据获得的bundle信息查询归属的Broker
                    return findRedirectLookupResultAsync(bundle).thenCompose(optResult -> {
                      			//如果findRedirectLookupResultAsync方式没查到则走这里进行查询
                            return findBrokerServiceUrl(bundle, options); 
                    });
                });

        future.thenAccept(optResult -> {
            ....
        }).exceptionally(ex -> {
						....
        });

        return future;
    }

先看看是怎么获取Topic所归属的Bundle的吧，就从getBundleAsync方法跟踪进去

    public CompletableFuture<NamespaceBundle> getBundleAsync(TopicName topic) {
        return bundleFactory.getBundlesAsync(topic.getNamespaceObject())
          			//直接看findBundle，命名意思已经很清晰了
                .thenApply(bundles -> bundles.findBundle(topic));
    }

    public NamespaceBundle findBundle(TopicName topicName) {
        checkArgument(nsname.equals(topicName.getNamespaceObject()));
      	//同理，继续跟踪进去
        return factory.getTopicBundleAssignmentStrategy().findBundle(topicName, this);
    }

    public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
        //计算Topic名称的哈希值
        long hashCode = Hashing.crc32().hashString(topicName.toString(), StandardCharsets.UTF_8).padToLong();
        //根据哈希值来获取所归属的bundle，一致性哈希的设计。跟进去看看是怎么计算的
        NamespaceBundle bundle = namespaceBundles.getBundle(hashCode);
        if (topicName.getDomain().equals(TopicDomain.non_persistent)) {
            bundle.setHasNonPersistentTopic(true);
        }
        return bundle;
    }

    protected NamespaceBundle getBundle(long hash) {
        //通过数组的二分查找进行计算，数组的元素个数跟存储Bundle的bundles的集合大小是一样的，能获取对应的Bundle
      	//思路其实就是一致性哈希的查找方式，计算出哈希值处于哈希环所处的位置并查找其下一个节点的信息
        int idx = Arrays.binarySearch(partitions, hash);
        int lowerIdx = idx < 0 ? -(idx + 2) : idx;
        return bundles.get(lowerIdx);
    }

知道Bundle之后，下一步就是根据这个Bundle来查询其所归属的Broker节点，也就是上面的NamespaceService类的findRedirectLookupResultAsync方法，这里一路跟下去就是查询缓存中获取映射信息的地方了，感兴趣的伙伴可以继续跟下去

    private CompletableFuture<Optional<LookupResult>> findRedirectLookupResultAsync(ServiceUnitId bundle) {
        if (isSLAOrHeartbeatNamespace(bundle.getNamespaceObject().toString())) {
            return CompletableFuture.completedFuture(Optional.empty());
        }
        return redirectManager.findRedirectLookupResultAsync();
    }

总结

以上就是Pulsar的Lookup机制的实现流程，在寻址的过程中，需要阅读的伙伴具备一致性哈希的知识，因为Pulsar的Topic归属就是引入了一致性哈希算法来实现的。


http://lihuaxi.xjx100.cn/news/2144103.html

相关文章

网络层IP协议和数据链路层--理解NAT/NAPT路由技术

网络层IP协议和数据链路层--理解NAT/NAPT路由技术

文章目录 一.网络层与数据链路层二.IP协议报头报文切片IP协议与路由器IP地址的结构与子网掩码数据报文路由的实现机制 三.公网IP和私网IP公网IP主机与公网IP主机之间的跨网络通信 四.数据链路层与局域网通信数据链路层以太网协议 五.NAT技术实现私网IP主机与公网IP主机之间的跨…
阅读更多...
Vue.js---------Vue基础

Vue.js---------Vue基础

能够说出Vue的概念和作用能够使用vue/cli脚手架工程化开发能够熟练Vue指令 一.vue基本概念 1.学习vue Vuex 是一个专为 Vue.js 应用程序开发的状态管理模式。它采用集中式存储管理应用的所有组件的状态&#xff0c;并以相应的规则保证状态以一种可预测的方式发生变化。 渐进…
阅读更多...
Apache ECharts-数据统计（详解、入门案例）

Apache ECharts-数据统计（详解、入门案例）

简介&#xff1a;Apache ECharts 是一款基于 Javascript 的数据可视化图表库&#xff0c;提供直观&#xff0c;生动&#xff0c;可交互&#xff0c;可个性化定制的数据可视化图表。 1、介绍 图 1.1 Apache ECharts 功能、运行环境 功能&#xff1a; ECharts&#xff…
阅读更多...
【opencv】教程代码 —Histograms_Matching（2）计算直方图、直方图比较、直方图均衡、模板匹配...

【opencv】教程代码 —Histograms_Matching（2）计算直方图、直方图比较、直方图均衡、模板匹配...

计算直方图直方图比较图像进行直方图均衡化处理模板匹配 1. calcHist_Demo.cpp 计算直方图 这段代码的功能是加载图像&#xff0c;分离图像的三个颜色通道&#xff0c;然后分别计算这三个通道的直方图&#xff0c;绘制出来并显示结果。直方图是图像中像素值分布的图形表示&…
阅读更多...
Docker实战教程 第2章 Docker基础

Docker实战教程 第2章 Docker基础

3-1 Docker介绍 什么是Docker 虚拟化&#xff0c;容器 Docker 是一个开源的应用容器引擎&#xff0c;基于 Go 语言 并遵从Apache2.0协议开源。 Docker 可以让开发者打包他们的应用以及依赖包到一个轻量级、可移植的容器中&#xff0c;然后发布到任何流行的 Linux 机器上&…
阅读更多...
丰诺畅机电科技将莅临2024年第13届生物发酵展

丰诺畅机电科技将莅临2024年第13届生物发酵展

参展企业介绍 无锡丰诺畅机电科技有限公司&#xff0c;是一家分离设备专业制造公司&#xff0c;集开发、设计、制造、销售、服务于一体;具有专业的生产技术&#xff0c;先进的生产工艺&#xff0c;精良的制造设备&#xff0c;完善的检测手段;为满足不同用户的过滤需求&#xf…
阅读更多...
三维VR虚拟展馆打破传统展览的时间与空间限制

三维VR虚拟展馆打破传统展览的时间与空间限制

探索绿色未来&#xff0c;见证生态转型——这是长江经济带在国家发展蓝图中的承诺。如今&#xff0c;通过线上长江经济带发展阶段性成效展厅&#xff0c;这一承诺正以创新的互动体验呈现给公众&#xff0c;彰显了环境保护与经济增长的和谐统一。 深圳VR公司华锐视点精心策划的长…
阅读更多...
easyExcel 模版导出 中间数据纵向延伸,并且对指定列进行合并

easyExcel 模版导出 中间数据纵向延伸,并且对指定列进行合并

想要达到的效果 引入maven引用 <dependency><groupId>com.alibaba</groupId><artifactId>easyexcel</artifactId><version>3.2.1</version></dependency> 按照要求创建模版 备注 : 模板注意 用{} 来表示你要用的变量 如果本…
阅读更多...
最新文章

Copyright @ 2022~2023