前段时间业务团队偶尔会碰到一些 Pulsar 使用的问题，比如消息阻塞不消费了、生产者消息发送缓慢等各种问题。
虽然我们有个监控页面可以根据 topic 维度查看他的发送状态，比如速率、流量、消费状态等信息。
本文重点不在于此，具体代码就不仔细分析了。
从上图可以看出，如果我们想要监控消费是否存在阻塞的情况，这几个内部队列是需要重点监控的，一旦他们出现堆积，那就会出现消费阻塞。
其实这些数据都可以通过。
org.apache.pulsar.client.api.ProducerStatsorg.apache.pulsar.client.api.ConsumerStats这两个接口获取到生产者和消费者的大部分指标，只是这里还有一个小插曲。
那就是在获取消费者队列大小的时候，获取到的数据一直为空。
最终经过源码排查，原来是我们大量使用的 messageListener 在获取队列大小时有 bug，导致获取到的数据一直都为 0.
相关的 issue 和 PR 可以在这两个链接查看，问题原因和修复过程都有具体描述：https://github.com/apache/pulsar/issues/20076 https://github.com/apache/pulsar/pull/20245
但这个修复得在新版本才能使用，就导致我们现在的监控页面一直显示为空。
然后就是开发一个 SkyWalking 的插件了，其实直接使用 SW 开发插件是上手 Java-Agent 比较快的方式。
SW 的 SDK 封装了许多 agent 原生接口，使得开发起来非常容易；当然缺点也有，就是得集成整个 SW 的 agent。
这里我简单介绍下这个插件的运行流程：
当消费多分区 topic 时，为了能唯一标志一个 consumer，所以给每个消费者都加了一个 hashcode 的 label。
因为我们所有的 Java 技术栈都是使用的 Prometheus 的包来生成 metrics ，所以该插件也是使用该包生成的数据。
<dependency> <groupId>io.prometheus</groupId> <artifactId>simpleclient</artifactId> <version>0.12.0</version> <scope>provided</scope> </dependency>为了兼容一些特殊 Java 应用没有该包时会启动报错，所以在初始化插件的时候需要检测当前 classpath 下是否存在该依赖。
这些功能 SW 已经封装好了，对我们来说也是开箱即用。
其实 SW 插件自己也是支持 metrics 的，由于我们只是使用了它的 trace 功能，所以这里就没有使用它的 API。
关于开发一个 SW 插件的流程也比较简单，可以参考官方文档或者是一些现成的插件源码。https://skywalking.apache.org/docs/skywalking-java/next/en/setup/service-agent/java-agent/java-plugin-development-guide/
有了这个监控面板后，对于 Pulsar 客户端内部的一些运行情况就不再是黑盒了，还可以基于此做一些报警，比如消费堆积、发送延迟过大等。
当然仅仅只有这个面板依然是不够的，后续我们又开发了可以通过 messageId 查询它的整个生命周期，包括：
同时借助与 Pulsar-SQL 的能力，还能以列表的形式展示当前 topic 的消息列表。
当然在实现这两个功能的同时也踩了不少坑，提了几个 PR ，后面在抽时间做具体的分享。
本文链接：http://www.28at.com/showinfo-26-91-0.html从 Pulsar Client 的原理到它的监控面板
声明：本网页内容旨在传播知识，若有侵权等问题请及时与本网联系，我们将在第一时间删除处理。邮件：2376512515@qq.com
上一篇： 一年经验在二线城市面试后端的经验分享
下一篇： 三言两语说透设计模式的艺术-单例模式
Copyright © 2016-2023 天津谷骐科技有限公司 版权所有 sitemap.xml
违法及侵权请联系：2376512515@qq.com 津ICP备18001702号
津公网安备 12010102000574号