1、需求的诞生
前几天公司我们部门需要演示一个应用，应用依赖kafka的数据，但是kafka的数据来自其他部门的投递。
一些原因导致数据无法给到，导致我们部门的演示也很有问题，所以想做一个简单的kafka topic的监控，在没有数据的时候及时发现并找兄弟部门沟通
这里记录下原因，因为机房的带宽只有500M，其他部门在做一些视频录制的工作，导致带宽满了，往kafka生产数据的producer无法发送到。
2、kafka监测
kafka的检测有很多方案，但是因为我们在测试环境使用，讲究一个轻量级，所以直接写一个小程序监控就得了。
kafka的监控没搞过，但是用过Offset Explorer，Offset中可以看到总的消息数量，延续这个思路，只要这个数值有变化，证明又在投递。这是思路
2.1 加入依赖
创建一个Springboot的项目，这个没难度，也不说了
这里把所有的依赖都放这了，主要是kafka和等会要用的钉钉的sdk
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.aliyun</groupId>
<artifactId>alibaba-dingtalk-service-sdk</artifactId>
<version>2.0.0</version>
</dependency>2.2 看下配置文件
application.yml
spring:
kafka:
bootstrap-servers: kafka地址
consumer:
auto-offset-reset: earliest
group-id: test-xiangcai
dingding:
webHook: https://oapi.dingtalk.com/robot/send?access_token=xxx
topics: topic1,topic2
sign: sign2.3 写一个类获取topic的数据
@Component
public class KafkaTools {
private static KafkaTools _this;
@Autowired
private ConsumerFactory<Long, String> consumerFactory;
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
Consumer<Long, String> consumer;
@PostConstruct
public void init() {
_this = this;
_this.consumer = consumerFactory.createConsumer();
}
public static Long getTopicCount(String topic) {
List<PartitionInfo> partitionInfos = _this.kafkaTemplate.partitionsFor(topic);
long count = 0L;
for (PartitionInfo str : partitionInfos) {
TopicPartition topicPartition = new TopicPartition(topic, str.partition());
long logEndOffset = _this.consumer.endOffsets(List.of(topicPartition)).get(topicPartition);
_this.consumer.assign(List.of(topicPartition));
long currentOffset = _this.consumer.position(topicPartition);
count += (logEndOffset - currentOffset);
System.err.println( " topic " + topic + " 最后的offset is " + logEndOffset + " currentOffset " + currentOffset + " totalCount " + (logEndOffset - currentOffset));
}
return count;
}
}解释下重点代码
LOG-END-OFFSET ：每个分区当前最新生产的消息的位移值
CURRENT-OFFSET ：该消费者组当前最新消费消息的位移值
注：因为内部是使用的单机的kafka，并且也只有一个partition
3、发送到钉钉
想做一个提醒，最初的方案是发送的邮箱，没有养成看邮件的习惯，所以想要发送到微信或者钉钉，看了下方案，钉钉是最容易实现的，所以这里选择使用钉钉
3.1 钉钉webhook
创建一个钉钉群，点击右上角群设置，机器人，进入到机器人的设置界面，然后点击添加机器人，选择自定义机器人，进入界面就好
这里可以设置机器人的头像，机器人的名字，同时也展示webhook的地址，可以通过这个url 发送消息到群里。
这里选择的安全设置加签的方式，可以参考说明文档，等会代码展示
3.2 钉钉发送消息到群里
看下代码，这里设置的消息为文本，并且使用了at全体人
/**
* 发送消息到钉钉
*
* @param topic
*/
private void sendMsgToDingDing(String topic) {
try {
String url = makeUrl(dingDingConfig.getWebHook(), dingDingConfig.getSign());
DingTalkClient client = new DefaultDingTalkClient(url);
OapiRobotSendRequest request = new OapiRobotSendRequest();
request.setMsgtype("text");
OapiRobotSendRequest.Text text = new OapiRobotSendRequest.Text();
text.setContent("192.168.2.8 kafka " + topic + "超过2分钟没有数据，检查下");
OapiRobotSendRequest.At at = new OapiRobotSendRequest.At();
// isAtAll类型如果不为Boolean，请升级至最新SDK
at.setIsAtAll(true);
request.setAt(at);
request.setText(text);
client.execute(request);
} catch (Exception e) {
}
}
/**
* 组装url
*
* @param url
* @param secret
* @return
* @throws Exception
*/
public static String makeUrl(String url, String secret) throws Exception {
Long timestamp = System.currentTimeMillis();
String stringToSign = timestamp + "\n" + secret;
Mac mac = Mac.getInstance("HmacSHA256");
mac.init(new SecretKeySpec(secret.getBytes("UTF-8"), "HmacSHA256"));
byte[] signData = mac.doFinal(stringToSign.getBytes("UTF-8"));
String sign = URLEncoder.encode(new String(Base64.encodeBase64(signData)), "UTF-8");
return url + String.format("×tamp=%d&sign=%s", timestamp, sign);
}看下效果
4、定时器功能
定时器直接使用springboot 的 schedule，在application上添加 直接 @EnableScheduling
创建定时job，并且设置2分钟执行一次，两分钟发现没有数据更新就会发送消息到钉钉提醒
public static Map<String, Long> countMap = new HashMap<>();
// @Scheduled(cron = "0/5 * * * * ? ") //每5秒执行一次
@Scheduled(cron = "0 0/2 * * * ?") // 2分钟执行一次
public void execute() {
for (String topic : dingDingConfig.getTopics().split(",")) {
Long topicCount = KafkaTools.getTopicCount(topic);
Long orDefault = countMap.getOrDefault(topic, 0L);
if (orDefault.equals(topicCount)) {
if (System.currentTimeMillis() - preDingTime > 60 * 60 * 1000L) {
sendMsgToDingDing(topic);
preDingTime = System.currentTimeMillis();
}
} else {
countMap.put(topic, topicCount);
}
}
System.err.println("job 执行时间 " + new Date());
}这里做了一些处理
第一个点就是 在内存中保存数量，没有使用数据库，主要考虑到轻量级，在测试环境
第二个就是钉钉消息的发送，防止出现消息喷涌，这里限制了1小时才可以发送一次
5、打包部署
看下dockerfile
#项目所依赖的jdk镜像
FROM openjdk:11-jdk
RUN mkdir -p /app
WORKDIR /app
#设置时区
RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
#将maven构建好的jar添加到镜像中，第二个为别名
COPY start.sh start.sh
ADD target/kafkaMonitor.jar kafkaMonitor.jar
#镜像所执行的命令
ENTRYPOINT ["sh","start.sh"]看下启动脚本
echo "192.168.2.8 kafka.tyjt.com" >> /etc/hosts
java -jar kafkaMonitor.jar
echo 'finish start!!'很简单直接启动，这个程序主打就是一个轻便
5、总结
这只是一个小的需求，没有使用一些重量级的组件，尽可能的快速的解决问题。
赠人玫瑰，手留余香，感谢点赞，关注，评论。
