mirror of
https://github.com/zhwei820/learn.lianglianglee.com.git
synced 2025-09-24 20:26:41 +08:00
390 lines
28 KiB
HTML
390 lines
28 KiB
HTML
<!DOCTYPE html>
|
||
<!-- saved from url=(0046)https://kaiiiz.github.io/hexo-theme-book-demo/ -->
|
||
<html xmlns="http://www.w3.org/1999/xhtml">
|
||
<head>
|
||
<head>
|
||
<meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1.0, user-scalable=no">
|
||
<link rel="icon" href="/static/favicon.png">
|
||
<title>32 KafkaAdminClient：Kafka的运维利器.md.html</title>
|
||
<!-- Spectre.css framework -->
|
||
<link rel="stylesheet" href="/static/index.css">
|
||
<!-- theme css & js -->
|
||
<meta name="generator" content="Hexo 4.2.0">
|
||
</head>
|
||
<body>
|
||
<div class="book-container">
|
||
<div class="book-sidebar">
|
||
<div class="book-brand">
|
||
<a href="/">
|
||
<img src="/static/favicon.png">
|
||
<span>技术文章摘抄</span>
|
||
</a>
|
||
</div>
|
||
<div class="book-menu uncollapsible">
|
||
<ul class="uncollapsible">
|
||
<li><a href="/" class="current-tab">首页</a></li>
|
||
</ul>
|
||
<ul class="uncollapsible">
|
||
<li><a href="../">上一级</a></li>
|
||
</ul>
|
||
<ul class="uncollapsible">
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/00 开篇词 为什么要学习Kafka？.md.html">00 开篇词 为什么要学习Kafka？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/01 消息引擎系统ABC.md.html">01 消息引擎系统ABC</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/02 一篇文章带你快速搞定Kafka术语.md.html">02 一篇文章带你快速搞定Kafka术语</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/03 Kafka只是消息引擎系统吗？.md.html">03 Kafka只是消息引擎系统吗？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/04 我应该选择哪种Kafka？.md.html">04 我应该选择哪种Kafka？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/05 聊聊Kafka的版本号.md.html">05 聊聊Kafka的版本号</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/06 Kafka线上集群部署方案怎么做？.md.html">06 Kafka线上集群部署方案怎么做？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/07 最最最重要的集群参数配置（上）.md.html">07 最最最重要的集群参数配置（上）</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/08 最最最重要的集群参数配置（下）.md.html">08 最最最重要的集群参数配置（下）</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/09 生产者消息分区机制原理剖析.md.html">09 生产者消息分区机制原理剖析</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/10 生产者压缩算法面面观.md.html">10 生产者压缩算法面面观</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/11 无消息丢失配置怎么实现？.md.html">11 无消息丢失配置怎么实现？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/12 客户端都有哪些不常见但是很高级的功能？.md.html">12 客户端都有哪些不常见但是很高级的功能？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/13 Java生产者是如何管理TCP连接的？.md.html">13 Java生产者是如何管理TCP连接的？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/14 幂等生产者和事务生产者是一回事吗？.md.html">14 幂等生产者和事务生产者是一回事吗？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/15 消费者组到底是什么？.md.html">15 消费者组到底是什么？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/16 揭开神秘的“位移主题”面纱.md.html">16 揭开神秘的“位移主题”面纱</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/17 消费者组重平衡能避免吗？.md.html">17 消费者组重平衡能避免吗？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/18 Kafka中位移提交那些事儿.md.html">18 Kafka中位移提交那些事儿</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/19 CommitFailedException异常怎么处理？.md.html">19 CommitFailedException异常怎么处理？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/20 多线程开发消费者实例.md.html">20 多线程开发消费者实例</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/21 Java 消费者是如何管理TCP连接的.md.html">21 Java 消费者是如何管理TCP连接的</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/22 消费者组消费进度监控都怎么实现？.md.html">22 消费者组消费进度监控都怎么实现？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/23 Kafka副本机制详解.md.html">23 Kafka副本机制详解</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/24 请求是怎么被处理的？.md.html">24 请求是怎么被处理的？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/25 消费者组重平衡全流程解析.md.html">25 消费者组重平衡全流程解析</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/26 你一定不能错过的Kafka控制器.md.html">26 你一定不能错过的Kafka控制器</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/27 关于高水位和Leader Epoch的讨论.md.html">27 关于高水位和Leader Epoch的讨论</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/28 主题管理知多少.md.html">28 主题管理知多少</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/29 Kafka动态配置了解下？.md.html">29 Kafka动态配置了解下？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/30 怎么重设消费者组位移？.md.html">30 怎么重设消费者组位移？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/31 常见工具脚本大汇总.md.html">31 常见工具脚本大汇总</a>
|
||
</li>
|
||
<li>
|
||
<a class="current-tab" href="/专栏/Kafka核心技术与实战/32 KafkaAdminClient：Kafka的运维利器.md.html">32 KafkaAdminClient：Kafka的运维利器</a>
|
||
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/33 Kafka认证机制用哪家？.md.html">33 Kafka认证机制用哪家？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/34 云环境下的授权该怎么做？.md.html">34 云环境下的授权该怎么做？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/35 跨集群备份解决方案MirrorMaker.md.html">35 跨集群备份解决方案MirrorMaker</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/36 你应该怎么监控Kafka？.md.html">36 你应该怎么监控Kafka？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/37 主流的Kafka监控框架.md.html">37 主流的Kafka监控框架</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/38 调优Kafka，你做到了吗？.md.html">38 调优Kafka，你做到了吗？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/39 从0搭建基于Kafka的企业级实时日志流处理平台.md.html">39 从0搭建基于Kafka的企业级实时日志流处理平台</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/40 Kafka Streams与其他流处理平台的差异在哪里？.md.html">40 Kafka Streams与其他流处理平台的差异在哪里？</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/41 Kafka Streams DSL开发实例.md.html">41 Kafka Streams DSL开发实例</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/42 Kafka Streams在金融领域的应用.md.html">42 Kafka Streams在金融领域的应用</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/加餐 搭建开发环境、阅读源码方法、经典学习资料大揭秘.md.html">加餐 搭建开发环境、阅读源码方法、经典学习资料大揭秘</a>
|
||
</li>
|
||
<li>
|
||
<a href="/专栏/Kafka核心技术与实战/结束语 以梦为马，莫负韶华！.md.html">结束语 以梦为马，莫负韶华！</a>
|
||
</li>
|
||
</ul>
|
||
</div>
|
||
</div>
|
||
<div class="sidebar-toggle" onclick="sidebar_toggle()" onmouseover="add_inner()" onmouseleave="remove_inner()">
|
||
<div class="sidebar-toggle-inner"></div>
|
||
</div>
|
||
<script>
|
||
function add_inner() {
|
||
let inner = document.querySelector('.sidebar-toggle-inner')
|
||
inner.classList.add('show')
|
||
}
|
||
function remove_inner() {
|
||
let inner = document.querySelector('.sidebar-toggle-inner')
|
||
inner.classList.remove('show')
|
||
}
|
||
function sidebar_toggle() {
|
||
let sidebar_toggle = document.querySelector('.sidebar-toggle')
|
||
let sidebar = document.querySelector('.book-sidebar')
|
||
let content = document.querySelector('.off-canvas-content')
|
||
if (sidebar_toggle.classList.contains('extend')) { // show
|
||
sidebar_toggle.classList.remove('extend')
|
||
sidebar.classList.remove('hide')
|
||
content.classList.remove('extend')
|
||
} else { // hide
|
||
sidebar_toggle.classList.add('extend')
|
||
sidebar.classList.add('hide')
|
||
content.classList.add('extend')
|
||
}
|
||
}
|
||
function open_sidebar() {
|
||
let sidebar = document.querySelector('.book-sidebar')
|
||
let overlay = document.querySelector('.off-canvas-overlay')
|
||
sidebar.classList.add('show')
|
||
overlay.classList.add('show')
|
||
}
|
||
function hide_canvas() {
|
||
let sidebar = document.querySelector('.book-sidebar')
|
||
let overlay = document.querySelector('.off-canvas-overlay')
|
||
sidebar.classList.remove('show')
|
||
overlay.classList.remove('show')
|
||
}
|
||
</script>
|
||
<div class="off-canvas-content">
|
||
<div class="columns">
|
||
<div class="column col-12 col-lg-12">
|
||
<div class="book-navbar">
|
||
<!-- For Responsive Layout -->
|
||
<header class="navbar">
|
||
<section class="navbar-section">
|
||
<a onclick="open_sidebar()">
|
||
<i class="icon icon-menu"></i>
|
||
</a>
|
||
</section>
|
||
</header>
|
||
</div>
|
||
<div class="book-content" style="max-width: 960px; margin: 0 auto;
|
||
overflow-x: auto;
|
||
overflow-y: hidden;">
|
||
<div class="book-post">
|
||
<p id="tip" align="center"></p>
|
||
<div><h1>32 KafkaAdminClient：Kafka的运维利器</h1>
|
||
<p>你好，我是胡夕。今天我要和你分享的主题是：Kafka 的运维利器 KafkaAdminClient。</p>
|
||
<h2>引入原因</h2>
|
||
<p>在上一讲中，我向你介绍了 Kafka 自带的各种命令行脚本，这些脚本使用起来虽然方便，却有一些弊端。</p>
|
||
<p>首先，不论是 Windows 平台，还是 Linux 平台，命令行的脚本都只能运行在控制台上。如果你想要在应用程序、运维框架或是监控平台中集成它们，会非常得困难。</p>
|
||
<p>其次，这些命令行脚本很多都是通过连接 ZooKeeper 来提供服务的。目前，社区已经越来越不推荐任何工具直连 ZooKeeper 了，因为这会带来一些潜在的问题，比如这可能会绕过 Kafka 的安全设置。在专栏前面，我说过 kafka-topics 脚本连接 ZooKeeper 时，不会考虑 Kafka 设置的用户认证机制。也就是说，任何使用该脚本的用户，不论是否具有创建主题的权限，都能成功“跳过”权限检查，强行创建主题。这显然和 Kafka 运维人员配置权限的初衷背道而驰。</p>
|
||
<p>最后，运行这些脚本需要使用 Kafka 内部的类实现，也就是 Kafka<strong>服务器端</strong>的代码。实际上，社区还是希望用户只使用 Kafka<strong>客户端</strong>代码，通过现有的请求机制来运维管理集群。这样的话，所有运维操作都能纳入到统一的处理机制下，方便后面的功能演进。</p>
|
||
<p>基于这些原因，社区于 0.11 版本正式推出了 Java 客户端版的 AdminClient，并不断地在后续的版本中对它进行完善。我粗略地计算了一下，有关 AdminClient 的优化和更新的各种提案，社区中有十几个之多，而且贯穿各个大的版本，足见社区对 AdminClient 的重视。</p>
|
||
<p>值得注意的是，<strong>服务器端也有一个 AdminClient</strong>，包路径是 kafka.admin。这是之前的老运维工具类，提供的功能也比较有限，社区已经不再推荐使用它了。所以，我们最好统一使用客户端的 AdminClient。</p>
|
||
<h2>如何使用？</h2>
|
||
<p>下面，我们来看一下如何在应用程序中使用 AdminClient。我们在前面说过，它是 Java 客户端提供的工具。想要使用它的话，你需要在你的工程中显式地增加依赖。我以最新的 2.3 版本为例来进行一下展示。</p>
|
||
<p>如果你使用的是 Maven，需要增加以下依赖项：</p>
|
||
<pre><code><dependency>
|
||
<groupId>org.apache.kafka</groupId>
|
||
<artifactId>kafka-clients</artifactId>
|
||
<version>2.3.0</version>
|
||
</dependency>
|
||
</code></pre>
|
||
<p>如果你使用的是 Gradle，那么添加方法如下：</p>
|
||
<pre><code>compile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.3.0'
|
||
</code></pre>
|
||
<h2>功能</h2>
|
||
<p>鉴于社区还在不断地完善 AdminClient 的功能，所以你需要时刻关注不同版本的发布说明（Release Notes），看看是否有新的运维操作被加入进来。在最新的 2.3 版本中，AdminClient 提供的功能有 9 大类。</p>
|
||
<ol>
|
||
<li>主题管理：包括主题的创建、删除和查询。</li>
|
||
<li>权限管理：包括具体权限的配置与删除。</li>
|
||
<li>配置参数管理：包括 Kafka 各种资源的参数设置、详情查询。所谓的 Kafka 资源，主要有 Broker、主题、用户、Client-id 等。</li>
|
||
<li>副本日志管理：包括副本底层日志路径的变更和详情查询。</li>
|
||
<li>分区管理：即创建额外的主题分区。</li>
|
||
<li>消息删除：即删除指定位移之前的分区消息。</li>
|
||
<li>Delegation Token 管理：包括 Delegation Token 的创建、更新、过期和详情查询。</li>
|
||
<li>消费者组管理：包括消费者组的查询、位移查询和删除。</li>
|
||
<li>Preferred 领导者选举：推选指定主题分区的 Preferred Broker 为领导者。</li>
|
||
</ol>
|
||
<h2>工作原理</h2>
|
||
<p>在详细介绍 AdminClient 的主要功能之前，我们先简单了解一下 AdminClient 的工作原理。<strong>从设计上来看，AdminClient 是一个双线程的设计：前端主线程和后端 I/O 线程</strong>。前端线程负责将用户要执行的操作转换成对应的请求，然后再将请求发送到后端 I/O 线程的队列中；而后端 I/O 线程从队列中读取相应的请求，然后发送到对应的 Broker 节点上，之后把执行结果保存起来，以便等待前端线程的获取。</p>
|
||
<p>值得一提的是，AdminClient 在内部大量使用生产者 - 消费者模式将请求生成与处理解耦。我在下面这张图中大致描述了它的工作原理。</p>
|
||
<p><img src="assets/4b520345918d0429801589217270d1eb.png" alt="img" /></p>
|
||
<p>如图所示，前端主线程会创建名为 Call 的请求对象实例。该实例有两个主要的任务。</p>
|
||
<ol>
|
||
<li><strong>构建对应的请求对象</strong>。比如，如果要创建主题，那么就创建 CreateTopicsRequest；如果是查询消费者组位移，就创建 OffsetFetchRequest。</li>
|
||
<li><strong>指定响应的回调逻辑</strong>。比如从 Broker 端接收到 CreateTopicsResponse 之后要执行的动作。一旦创建好 Call 实例，前端主线程会将其放入到新请求队列（New Call Queue）中，此时，前端主线程的任务就算完成了。它只需要等待结果返回即可。</li>
|
||
</ol>
|
||
<p>剩下的所有事情就都是后端 I/O 线程的工作了。就像图中所展示的那样，该线程使用了 3 个队列来承载不同时期的请求对象，它们分别是新请求队列、待发送请求队列和处理中请求队列。为什么要使用 3 个呢？原因是目前新请求队列的线程安全是由 Java 的 monitor 锁来保证的。<strong>为了确保前端主线程不会因为 monitor 锁被阻塞，后端 I/O 线程会定期地将新请求队列中的所有 Call 实例全部搬移到待发送请求队列中进行处理</strong>。图中的待发送请求队列和处理中请求队列只由后端 I/O 线程处理，因此无需任何锁机制来保证线程安全。</p>
|
||
<p>当 I/O 线程在处理某个请求时，它会显式地将该请求保存在处理中请求队列。一旦处理完成，I/O 线程会自动地调用 Call 对象中的回调逻辑完成最后的处理。把这些都做完之后，I/O 线程会通知前端主线程说结果已经准备完毕，这样前端主线程能够及时获取到执行操作的结果。AdminClient 是使用 Java Object 对象的 wait 和 notify 实现的这种通知机制。</p>
|
||
<p>严格来说，AdminClient 并没有使用 Java 已有的队列去实现上面的请求队列，它是使用 ArrayList 和 HashMap 这样的简单容器类，再配以 monitor 锁来保证线程安全的。不过，鉴于它们充当的角色就是请求队列这样的主体，我还是坚持使用队列来指代它们了。</p>
|
||
<p>了解 AdminClient 工作原理的一个好处在于，<strong>它能够帮助我们有针对性地对调用 AdminClient 的程序进行调试</strong>。</p>
|
||
<p>我们刚刚提到的后端 I/O 线程其实是有名字的，名字的前缀是 kafka-admin-client-thread。有时候我们会发现，AdminClient 程序貌似在正常工作，但执行的操作没有返回结果，或者 hang 住了，现在你应该知道这可能是因为 I/O 线程出现问题导致的。如果你碰到了类似的问题，不妨使用<strong>jstack 命令</strong>去查看一下你的 AdminClient 程序，确认下 I/O 线程是否在正常工作。</p>
|
||
<p>这可不是我杜撰出来的好处，实际上，这是实实在在的社区 bug。出现这个问题的根本原因，就是 I/O 线程未捕获某些异常导致意外“挂”掉。由于 AdminClient 是双线程的设计，前端主线程不受任何影响，依然可以正常接收用户发送的命令请求，但此时程序已经不能正常工作了。</p>
|
||
<h2>构造和销毁 AdminClient 实例</h2>
|
||
<p>如果你正确地引入了 kafka-clients 依赖，那么你应该可以在编写 Java 程序时看到 AdminClient 对象。<strong>切记它的完整类路径是 org.apache.kafka.clients.admin.AdminClient，而不是 kafka.admin.AdminClient</strong>。后者就是我们刚才说的服务器端的 AdminClient，它已经不被推荐使用了。</p>
|
||
<p>创建 AdminClient 实例和创建 KafkaProducer 或 KafkaConsumer 实例的方法是类似的，你需要手动构造一个 Properties 对象或 Map 对象，然后传给对应的方法。社区专门为 AdminClient 提供了几十个专属参数，最常见而且必须要指定的参数，是我们熟知的<strong>bootstrap.servers 参数</strong>。如果你想了解完整的参数列表，可以去<a href="https://kafka.apache.org/documentation/#adminclientconfigs">官网</a>查询一下。如果要销毁 AdminClient 实例，需要显式调用 AdminClient 的 close 方法。</p>
|
||
<p>你可以简单使用下面的代码同时实现 AdminClient 实例的创建与销毁。</p>
|
||
<pre><code>Properties props = new Properties();
|
||
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-host:port");
|
||
props.put("request.timeout.ms", 600000);
|
||
|
||
try (AdminClient client = AdminClient.create(props)) {
|
||
// 执行你要做的操作……
|
||
}
|
||
</code></pre>
|
||
<p>这段代码使用 Java 7 的 try-with-resource 语法特性创建了 AdminClient 实例，并在使用之后自动关闭。你可以在 try 代码块中加入你想要执行的操作逻辑。</p>
|
||
<h2>常见的 AdminClient 应用实例</h2>
|
||
<p>讲完了 AdminClient 的工作原理和构造方法，接下来，我举几个实际的代码程序来说明一下如何应用它。这几个例子，都是我们最常见的。</p>
|
||
<h3>创建主题</h3>
|
||
<p>首先，我们来看看如何创建主题，代码如下：</p>
|
||
<pre><code>String newTopicName = "test-topic";
|
||
try (AdminClient client = AdminClient.create(props)) {
|
||
NewTopic newTopic = new NewTopic(newTopicName, 10, (short) 3);
|
||
CreateTopicsResult result = client.createTopics(Arrays.asList(newTopic));
|
||
result.all().get(10, TimeUnit.SECONDS);
|
||
}
|
||
</code></pre>
|
||
<p>这段代码调用 AdminClient 的 createTopics 方法创建对应的主题。构造主题的类是 NewTopic 类，它接收主题名称、分区数和副本数三个字段。</p>
|
||
<p>注意这段代码倒数第二行获取结果的方法。目前，AdminClient 各个方法的返回类型都是名为 ***Result 的对象。这类对象会将结果以 Java Future 的形式封装起来。如果要获取运行结果，你需要调用相应的方法来获取对应的 Future 对象，然后再调用相应的 get 方法来取得执行结果。</p>
|
||
<p>当然，对于创建主题而言，一旦主题被成功创建，任务也就完成了，它返回的结果也就不重要了，只要没有抛出异常就行。</p>
|
||
<h3>查询消费者组位移</h3>
|
||
<p>接下来，我来演示一下如何查询指定消费者组的位移信息，代码如下：</p>
|
||
<pre><code>String groupID = "test-group";
|
||
try (AdminClient client = AdminClient.create(props)) {
|
||
ListConsumerGroupOffsetsResult result = client.listConsumerGroupOffsets(groupID);
|
||
Map<TopicPartition, OffsetAndMetadata> offsets =
|
||
result.partitionsToOffsetAndMetadata().get(10, TimeUnit.SECONDS);
|
||
System.out.println(offsets);
|
||
}
|
||
</code></pre>
|
||
<p>和创建主题的风格一样，<strong>我们调用 AdminClient 的 listConsumerGroupOffsets 方法去获取指定消费者组的位移数据</strong>。</p>
|
||
<p>不过，对于这次返回的结果，我们不能再丢弃不管了，<strong>因为它返回的 Map 对象中保存着按照分区分组的位移数据</strong>。你可以调用 OffsetAndMetadata 对象的 offset() 方法拿到实际的位移数据。</p>
|
||
<h3>获取 Broker 磁盘占用</h3>
|
||
<p>现在，我们来使用 AdminClient 实现一个稍微高级一点的功能：获取某台 Broker 上 Kafka 主题占用的磁盘空间量。有些遗憾的是，目前 Kafka 的 JMX 监控指标没有提供这样的功能，而磁盘占用这件事，是很多 Kafka 运维人员要实时监控并且极为重视的。</p>
|
||
<p>幸运的是，我们可以使用 AdminClient 来实现这一功能。代码如下：</p>
|
||
<pre><code>try (AdminClient client = AdminClient.create(props)) {
|
||
DescribeLogDirsResult ret = client.describeLogDirs(Collections.singletonList(targetBrokerId)); // 指定 Broker id
|
||
long size = 0L;
|
||
for (Map<String, DescribeLogDirsResponse.LogDirInfo> logDirInfoMap : ret.all().get().values()) {
|
||
size += logDirInfoMap.values().stream().map(logDirInfo -> logDirInfo.replicaInfos).flatMap(
|
||
topicPartitionReplicaInfoMap ->
|
||
topicPartitionReplicaInfoMap.values().stream().map(replicaInfo -> replicaInfo.size))
|
||
.mapToLong(Long::longValue).sum();
|
||
}
|
||
System.out.println(size);
|
||
}
|
||
</code></pre>
|
||
<p>这段代码的主要思想是，使用 AdminClient 的<strong>describeLogDirs 方法</strong>获取指定 Broker 上所有分区主题的日志路径信息，然后把它们累积在一起，得出总的磁盘占用量。</p>
|
||
<h2>小结</h2>
|
||
<p>好了，我们来小结一下。社区于 0.11 版本正式推出了 Java 客户端版的 AdminClient 工具，该工具提供了几十种运维操作，而且它还在不断地演进着。如果可以的话，你最好统一使用 AdminClient 来执行各种 Kafka 集群管理操作，摒弃掉连接 ZooKeeper 的那些工具。另外，我建议你时刻关注该工具的功能完善情况，毕竟，目前社区对 AdminClient 的变更频率很高。</p>
|
||
<p><img src="assets/34082877019e87bd30320789e96e6b23.jpg" alt="img" /></p>
|
||
</div>
|
||
</div>
|
||
<div>
|
||
<div style="float: left">
|
||
<a href="/专栏/Kafka核心技术与实战/31 常见工具脚本大汇总.md.html">上一页</a>
|
||
</div>
|
||
<div style="float: right">
|
||
<a href="/专栏/Kafka核心技术与实战/33 Kafka认证机制用哪家？.md.html">下一页</a>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
</div>
|
||
<a class="off-canvas-overlay" onclick="hide_canvas()"></a>
|
||
</div>
|
||
<script defer src="https://static.cloudflareinsights.com/beacon.min.js/v652eace1692a40cfa3763df669d7439c1639079717194" integrity="sha512-Gi7xpJR8tSkrpF7aordPZQlW2DLtzUlZcumS8dMQjwDHEnw9I7ZLyiOj/6tZStRBGtGgN6ceN6cMH8z7etPGlw==" data-cf-beacon='{"rayId":"709972135e213d60","version":"2021.12.0","r":1,"token":"1f5d475227ce4f0089a7cff1ab17c0f5","si":100}' crossorigin="anonymous"></script>
|
||
</body>
|
||
<!-- Global site tag (gtag.js) - Google Analytics -->
|
||
<script async src="https://www.googletagmanager.com/gtag/js?id=G-NPSEEVD756"></script>
|
||
<script>
|
||
window.dataLayer = window.dataLayer || [];
|
||
function gtag() {
|
||
dataLayer.push(arguments);
|
||
}
|
||
gtag('js', new Date());
|
||
gtag('config', 'G-NPSEEVD756');
|
||
var path = window.location.pathname
|
||
var cookie = getCookie("lastPath");
|
||
console.log(path)
|
||
if (path.replace("/", "") === "") {
|
||
if (cookie.replace("/", "") !== "") {
|
||
console.log(cookie)
|
||
document.getElementById("tip").innerHTML = "<a href='" + cookie + "'>跳转到上次进度</a>"
|
||
}
|
||
} else {
|
||
setCookie("lastPath", path)
|
||
}
|
||
function setCookie(cname, cvalue) {
|
||
var d = new Date();
|
||
d.setTime(d.getTime() + (180 * 24 * 60 * 60 * 1000));
|
||
var expires = "expires=" + d.toGMTString();
|
||
document.cookie = cname + "=" + cvalue + "; " + expires + ";path = /";
|
||
}
|
||
function getCookie(cname) {
|
||
var name = cname + "=";
|
||
var ca = document.cookie.split(';');
|
||
for (var i = 0; i < ca.length; i++) {
|
||
var c = ca[i].trim();
|
||
if (c.indexOf(name) === 0) return c.substring(name.length, c.length);
|
||
}
|
||
return "";
|
||
}
|
||
</script>
|
||
</html>
|