Apache Pulsar源码解析之Lookup机制

引言

在学习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所归属的BundleCompletableFuture<Optional<LookupResult>> future = getBundleAsync(topic).thenCompose(bundle -> {//根据获得的bundle信息查询归属的Brokerreturn 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://www.mzph.cn/news/790679.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

针孔相机、鱼眼相机、全景相机

针孔相机、鱼眼相机、全景相机

先进性简述&#xff0c;后续慢慢会补充1. 针孔相机&#xff1a; 针孔相机是一种基于针孔成像原理的传统相机&#xff0c;它使用一个非常小的孔径&#xff08;即“针孔”&#xff09;来限制光线进入相机的方式。 这种相机通常具有简单的结构&#xff0c;由一个孔径较小的光学元…
阅读更多...
acwing算法提高之图论--最小生成树的扩展应用

acwing算法提高之图论--最小生成树的扩展应用

目录 1 介绍2 训练 1 介绍 本专题用来记录使用最小生成树算法&#xff08;prim或kruskal&#xff09;解决的扩展题目。 2 训练 题目1&#xff1a;1146新的开始 C代码如下&#xff0c; #include <iostream> #include <cstring> #include <algorithm>usin…
阅读更多...
Flutter中setState函数的使用注意事项

Flutter中setState函数的使用注意事项

文章目录 Flutter中setState函数的使用注意事项只能在具有State对象的类中使用不要在build方法中使用将状态更新逻辑放在setState方法内部避免频繁调用使用回调函数更新状态 Flutter中setState函数的使用注意事项 setState()函数是Flutter中非常重要的一个函数&#xff0c;它用…
阅读更多...
C#使用Selenium驱动Chrome浏览器

C#使用Selenium驱动Chrome浏览器

1.Selenium库依赖安装 Selenium WebDriver是Selenium项目的一部分&#xff0c;用于模拟用户在Web应用程序中的交互操作。它支持多种浏览器&#xff0c;如Chrome、Firefox、IE等&#xff0c;且与各种编程语言&#xff08;如Java、Python、C#等&#xff09;兼容&#xff0c;具有…
阅读更多...
运用抽象工厂模式实现多个产品族的独立创建与组合管理

运用抽象工厂模式实现多个产品族的独立创建与组合管理

抽象工厂模式&#xff08;Abstract Factory Pattern&#xff09; 是一种软件设计模式&#xff0c;属于创建型设计模式类别&#xff0c;主要用于处理对象的创建过程&#xff0c;特别是在应对具有多个相关或相互依赖对象的特定上下文时。该模式的主要目的是将一组相关或者相互依赖…
阅读更多...
SpringMvc处理器方法的返回值

SpringMvc处理器方法的返回值

1、返回ModelAndView ModelAndView中的视图会经过你视图解析器处理得到真实的jsp文件页面并且跳转。 RequestMapping("test01")public ModelAndView test01() {ModelAndView mv new ModelAndView();// 携带数据mv.addObject("teamName", "牛逼&quo…
阅读更多...
AI绘图:Controlnet在ComfyUI中的使用教程：提升Stable Diffusion可控性

AI绘图:Controlnet在ComfyUI中的使用教程：提升Stable Diffusion可控性

前言 Controlnet是Stable Diffusion中一个至关重要的功能&#xff0c;它极大地增强了图像生成的可控性。通过引入额外的条件&#xff0c;Controlnet使我们能够更精确地控制生成的图像内容。本教程将详细介绍如何在ComfyUI中使用Controlnet&#xff0c;从而让你的创作过程更加灵…
阅读更多...
2012年认证杯SPSSPRO杯数学建模B题(第二阶段)节能减排全过程文档及程序

2012年认证杯SPSSPRO杯数学建模B题(第二阶段)节能减排全过程文档及程序

2012年认证杯SPSSPRO杯数学建模 节能减排、抑制全球气候变暖 B题 白屋顶计划 原题再现&#xff1a; 第二阶段问题   虽然环境学家对地球环境温度的改变有许多种不同观点&#xff0c;但大多数科学家可以达成一个基本的共识&#xff1a;近年来人类的活动&#xff0c;尤指二氧…
阅读更多...
云存储中常用的相同子策略的高效、安全的基于属性的访问控制的论文阅读

云存储中常用的相同子策略的高效、安全的基于属性的访问控制的论文阅读

参考文献为2022年发表的Efficient and Secure Attribute-Based Access Control With Identical Sub-Policies Frequently Used in Cloud Storage 动机 ABE是实现在云存储中一种很好的访问控制手段&#xff0c;但是其本身的计算开销导致在实际场景中应用收到限制。本论文研究了…
阅读更多...
ESP32学习---ESP-NOW（一）

ESP32学习---ESP-NOW（一）

ESP32学习---ESP-NOW&#xff08;一&#xff09; 官网简介arduino 官网简介 首先看官网的介绍&#xff1a;https://www.espressif.com.cn/zh-hans/solutions/low-power-solutions/esp-now ESP-NOW 是乐鑫定义的一种无线通信协议&#xff0c;能够在无路由器的情况下直接、快速…
阅读更多...
【Java多线程（4）】案例：设计模式

【Java多线程（4）】案例：设计模式

目录 一、什么是设计模式&#xff1f; 二、单例模式 1. 饿汉模式 2. 懒汉模式 懒汉模式-第一次改进 懒汉模式-第二次改进 懒汉模式-第三次改进 一、什么是设计模式&#xff1f; 设计模式是针对软件设计中常见问题的通用解决方案。它们提供了一种被广泛接受的方法来解决…
阅读更多...
5米分辨率数字高程模型（DEM）的制作

5米分辨率数字高程模型（DEM）的制作

在现代科技的驱动下&#xff0c;地理信息系统&#xff08;GIS&#xff09;和遥感技术已经取得了惊人的进展。其中一项令人瞩目的技术就是5米分辨率数字高程模型&#xff08;DEM&#xff09;的制作&#xff0c;它是基于多颗高分辨率卫星数据为原始数据&#xff0c;借助智能立体模…
阅读更多...
C语言编写Linux的Shell外壳

C语言编写Linux的Shell外壳

目录 一、输出命令行 1.1 了解环境变量 1.2 获取用户名、主机名、当前路径 1.3 缓冲区改进MakeCommandLine 二、获取用户命令 2.1 读取函数的选择 2.2 细节优化 2.3 返回值 三、指令和选项分割 3.1 strtok 函数 3.2 分割实现 四、执行命令 4.1 fork 方法 4.2 进…
阅读更多...
0.17元的4位数码管驱动芯片AiP650，支持键盘，还是无锡国家集成电路设计中心某公司的

0.17元的4位数码管驱动芯片AiP650，支持键盘，还是无锡国家集成电路设计中心某公司的

推荐原因&#xff1a;便宜的4位数码管驱动芯片 只要0.17元&#xff0c;香吗&#xff1f;X背景的哦。 2 线串口共阴极 8 段 4 位 LED 驱动控制/7*4 位键盘扫描专用电路 AIP650参考电路图 AIP650引脚定义
阅读更多...
scratch小动物的晚会 2024年3月中国电子学会图形化编程 少儿编程 scratch编程等级考试一级真题和答案解析

scratch小动物的晚会 2024年3月中国电子学会图形化编程 少儿编程 scratch编程等级考试一级真题和答案解析

目录 scratch小动物的晚会 一、题目要求 1、准备工作 2、功能实现 二、案例分析
阅读更多...
51单片机入门_江协科技_20.1_Proteus串口仿真

51单片机入门_江协科技_20.1_Proteus串口仿真

1.为了解决51单片机学习过程中在Proteus中的串口仿真的问题&#xff0c;需要在Proteus中建立串口仿真的环境&#xff08;目前Proteus安装在Win7x64虚拟机环境中&#xff1b; 2. 在CSDN中找到VSPD下载地址&#xff0c;在虚拟机中进行VSPD的安装&#xff0c;具体链接地址如下&am…
阅读更多...
mac电脑下pip安装库后，仍然提示command not found

mac电脑下pip安装库后，仍然提示command not found

如&#xff1a;pip install httprunner&#xff0c;安装成功后输入hrun -V 提示“command not found”。 造成这个问题的原因是PATH环境变量的问题&#xff0c;因为使用的是homebrew安装的python&#xff0c;导致pip bin目录的环境变量没注册上。 解决办法是将python bin目录加…
阅读更多...
HTML块级元素和内联元素（头部和布局）

HTML块级元素和内联元素（头部和布局）

目录 1.HTML块级和内联标签&#xff1a; 1.块级元素&#xff1a; 2.内联元素: 3.元素嵌套&#xff1a; 4.元素转换&#xff1a; 示例如下: 2.内联框架&#xff1a; 前言&#xff1a; 示例如下: 3.布局&#xff1a; 4.头部标签&#xff1a; 前言&#xff1a; 说明&…
阅读更多...
【Elasticsearch配置秘籍】详解config/elasticsearch.yml，解锁关键参数与实战应用

【Elasticsearch配置秘籍】详解config/elasticsearch.yml，解锁关键参数与实战应用

Elasticsearch的强大性能与灵活性&#xff0c;很大程度上得益于其细致入微的配置选项。作为核心配置文件&#xff0c;config/elasticsearch.yml扮演着至关重要的角色。本文将深度剖析该文件中的重要参数&#xff0c;结合实例说明其作用与应用场景&#xff0c;助您定制最适合业务…
阅读更多...
【蓝桥杯-Even Parity】

【蓝桥杯-Even Parity】

蓝桥杯-Even Parity 洛谷 UVA11464 Even Parity 暴力思路&#xff1a; 去遍历每个元素&#xff0c;如果不符合要求则翻转 时间复杂度大概在O&#xff08;2^&#xff08;nn&#xff09; nn&#xff09; 改进思路&#xff1a; 先去枚举确定第一行&#xff08;第一行得合法&…
阅读更多...
最新文章

Copyright @ 2022~2023