网管系统展示并管理着部分网元的数据和业务，当网元数据发生变更时，会通过不同的网管协议向 网管发送消息，网管接入层解析这些消息后会转换为对应的Kafka消息。（5G网元与网管通过 NETCONF 协议通信，4G网元通过 SNMP 协议通信）。对于消息的处理，网络自优化功能主要处理的消息分为两种
1、直接解析入库展示。重点是处理大批量数据的 kafka 接收、多线程并发处理、elasticsearch 入库问题
2、 根据消息内容进行业务处理。重点是网元在不同场景下触发的业务消息，对网元的数据进行增删改查等操作和对一些小区数据进行 KPI 计算和判断

我的工作
1. 实现了从 Kafka 拉取并处理消息、创建索引的架构
2. 通过修改 ES 索引存储和清理策略，优化消息处理效率，消除了 Kafka 的消息积压：原先一中消息类型的数据存储于一个索引中，峰值时期每天的数据量大约在300~500G左右，需要及时清理。在定时清理索引时，采用 deleteByQuery 方式，删除效率很低。因此采用索引模板+滚动索引的方式，以时间为单位创建索引。删除时，直接删除超出时间范围的索引。
3. 优化了极端场景下 ES 数据查询异常的问题：经分析，是由于并发查询数据较大，导致通过 NIO 读取到直接内存时，设置的直接内存大小不足。通过计算峰值数据大小，修改了 JVM 参数 MaxDirectMemorySize。