Zabbix trapper脚本推送监控数据

IT那活儿 2024-02-26

674

点击上方“IT那活儿”公众号--专注于企业全栈运维技术分享，不管IT什么活儿，干就完了！！！

前言

Zabbix是一种企业级的开源网络监控解决方案。它是一个全功能的、开源的解决方案，可以监控系统中大部分数据。

Zabbix中有多达十三种数据采集方法，每种方法所使用的原理和场景也不一样。自动化运维平台中最常用的采集方式是通过脚本采集数据，根据监控配置周期性的执行的脚本程序或命令，然后zaibbix主动采集脚本输出的内容作为监控项数据。

本文讲述通过Zabbix Trapper方式，被监控的设备或系统主动向Zabbix服务器或Zabbix代理发送数据。这种方式非常灵活，可以根据实际需要随时发送数据，而不需要等待Zabbix服务器或Zabbix代理的轮询。通常用于处理那些不适合被动采集，但又需要实时监控的数据。

案例描述

监听kafka端口，当kafka中吐出消息是，我们利用自发现规则创建监控项，并向监控项推送数据。由于kafka消息并不是系统中可以随时采集到的参数，这种情况通过zabbix主动采集数据的方式已经行不通，需要利用Zabbix Trapper方式被动的收集数据。

2.1 引入依赖

创建一个Springboot项目，在pom.xml中引入kafka和zabbix的依赖。

<dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
        <groupId>io.github.hengyunabc</groupId>
        <artifactId>zabbix-sender</artifactId>
        <version>0.0.5</version>
</dependency>

2.2 编辑配置

编辑application.yml配置文件，配置kafka连接信息，包括kafka IP、端口、topic、zabbix IP、端口、监控项绑定的主机。

kafka:
    consumer:
      enable-auto-commit: true
      group-id: ${spring.application.name}
      #消费模式 
      #latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 
      #earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      auto-offset-reset: earliest
      bootstrap-servers: 127.0.xx.xx:xxxx,xx.xx.xx.xx:xxxx #kafka地址和端口
      max-poll-records: 20 # 一次 poll 最多返回的记录数
    listener:
      type: batch # 开启批量消费
#消费的topic 
kafka:
  zabbix:
    topic: xx-topic-xx 

# zabbix信息 
zabbix: 
  hostname: xx.xx.xx.xx
  port: xxxx
# 监控项绑定的主机
binding:
  hostname: xx.xx.xx.xx

2.3 加载配置

创建一个类，利用 @Component 该类其标识为Spring的一个组件，利用 @Value 注解读取配置文件中的属性值并将其，注入到 Bean 对象的对应属性中。

@Component
public class ZabbixConsumer{ 
    @Value("${zabbix.hostname}")
    private String hostName;

    @Value("${zabbix.port}")
    private int port;

    @Value("${binding.hostname}")
    private String bindingHostName;
}

在ZabbixConsumer中声明一个ZabbixSender发送器，利用单例模式，避免重复创建消耗资源。

这里使用的是懒汉单例，懒汉单例和饿汉单例是两种常见的单例设计模式，它们的主要区别在于实例化对象的时机和线程安全性。

懒汉单例
在被外部类调用时创建实例，特点是延迟加载，即在需要的时候才创建对象。如果在创建实例对象时不加上synchronized关键字，则会导致对对象的访问不是线程安全的。但是，如果加上synchronized关键字，虽然可以保证线程安全，但是效率会降低。
饿汉单例
在类被加载时就会实例化一个对象，特点是在类加载的时候就完成实例化，避免了线程同步问题。但是如果从未使用过这个实例，则会造成内存的浪费。

private ZabbixSender sender = null; 

public ZabbixSender getZabbixSender() {
        if(sender == null) {
            sender = new ZabbixSender(hostName, port);
        }
        return sender;
}

2.4 Kafka消息监听器

创建一个receive方法，利用 @KafkaListener(topics = "#{'${kafka.zabbix.topic}'.split(',')}") 将方法声明为 Kafka 消息监听器，它会监听配置文件中 kafka.zabbix.topic 属性指定的一个或多个 Kafka topic。

EXECUTOR_SERVICE是一个线程池，它的核心线程数为10，最大线程数为500，空闲线程的存活时间为5秒，使用的阻塞队列是一个容量为500的ArrayBlockingQueue，当接收到消息时，使用线程池处理这些消息，可以有效地处理大量的并发消息，提高系统的吞吐量。

public static final ThreadPoolExecutor EXECUTOR_SERVICE = 
new ThreadPoolExecutor(10, 500, 5, TimeUnit.SECONDS,new ArrayBlockingQueue<>(500));

    @KafkaListener(topics = "#{'${kafka.zabbix.topic}'.split(',')}")
    public void receive(List<String> list) {
        try {
            EXECUTOR_SERVICE.execute(new ZabbixHBService(list,getZabbixSender(),bindingHostName));
        } catch (Exception e) {
            log.error("执行ZabbixHBService异常", e);
        }

    }

2.5 zabbix监控自发现和zabbix监控项数据

在监控项较多的时候，纯手工添加监控项的效率势必低下，所以我们利用 Zabbix 的自动发现代替手动的添加监控项，减轻工作量，每次向监控项推送数据前利用监控自发现创建相应监控项，然后向创建的监控中推送数据。

colock:当前时间的时间戳，Java中常用System.currentTimeMillis()。
bindingHostName：自发现和监控项推送到的主机，如 linux_127.0.0.1。
selfDiscoveryKey：自发现键值。
zabbixKey：监控项键值。
JSON.toJSONString(itemMap)：要推送的数据，推送时转换为json格式，方便阅读。

//监控项自发现
DataObject itemObject = new DataObject(clock, bindingHostName, selfDiscoveryKey, JSON.toJSONString(dataMap));
//监控项数据
DataObject dataObject = new DataObject(clock, bindingHostName, zabbixKey, JSON.toJSONString(itemMap));
/* 自发现监控项 批量处理推送zabbix */
if (itemObjectList.size() > 0) {
            try {
                log.info("自发现监控项 itemObjectList：{}", JSON.toJSONString(itemObjectList));
                SenderResult result = zabbixSender.send(itemObjectList);
                log.info("zabbixSender返回结果 result：{}", JSONObject.toJSONString(result));
            } catch (IOException e) {
                log.error("zabbixSender发送异常", e);
            }
 }
if (dataObjectList.size() > 0) {
            try {
                log.info("监控项数据 dataObjectList：{}", JSON.toJSONString(dataObjectList));
                SenderResult result = zabbixSender.send(dataObjectList);
                log.info("zabbixSender返回结果 result：{}", JSONObject.toJSONString(result));
            } catch (IOException e) {
                log.error("zabbixSender发送异常", e);
            }
}