服务监测与redis序列化

前往原站点查看

2023-01-09 00:17:45

    随着服务器数量的增多,部署于不同服务器的服务数量也是稳步增长的,这时候监测每个服务是否正常运转就显得尤为重要。通过监测的结果,可以方便的知道有没有服务下线了,从而采取相应的解决策略。

服务监测的设计

    我设计的监测策略如下:

    所有服务监测相关的数据存于redis中,每隔5分钟向目标服务端口发送一个socket连接请求,如果连接成功了那么确认服务存活,如果出现异常判定服务异常下线。对于每个服务,可以设置是否启用下线报警(发送邮件通知),如果启用,将在发现下线时,向管理员的邮箱发送一封邮件告知,并且暂且关闭该服务的报警功能(因为设置的监测时间比较密集,如果不暂且关闭的话,一直发送邮件会很快塞满邮箱)。

    存储的数据类型是redis的hash类型,对应的java类型为 Map<String, List<DetectedPort>> ,其中key是监测的服务器ip地址,value是一个list集合,该集合每个成员都是一个端口的相关信息(包括了端口号、服务名称、状态、是否警告)。当然因为可以对整个服务器进行监测,所以额外的增加一个hash类型的数据,对应的java类型为 Map<String, Boolean> ,其中key也是ip地址,value是该服务器是否在线。

    对于服务监测功能,一共有这样几个api的实现:增加服务器、删除服务器、设置端口、删除端口、更新状态、查询状态。

    具体的设计细节如下:

DetectedPort 类 :被监测的端口

@Data
@AllArgsConstructor
@NoArgsConstructor
public class DetectedPort {
    private Integer port;
    private String name;
    /**
     * 判断当前应用是否在线
     */
    private Boolean up;
    /**
     * 当服务下线时是否通知
     */
    private Boolean alert;
}

DetectedServer 类 : 被监测的服务器

@Data
@AllArgsConstructor
@NoArgsConstructor
public class DetectedServer {
    /**
     * 判断当前服务器是否在线
     */
    private String ip;
    private Boolean up;  // 一个服务器可以有多个端口号
    private List<DetectedPort> ports; 
}

DsapUtil 类 :监测工具类,用来监测服务和端口是否可以连接

public class DsapUtil {
    private static final int TIMEOUT = 500;

    public static boolean serverDetect(String host){
        try {
            return InetAddress.getByName(host).isReachable(TIMEOUT);
        } catch (Exception e) {
            return false;
        }
    }
    public static boolean portDetect(String host, int port){
        Socket socket = null;
        try {
            socket = new Socket();
            socket.connect(new InetSocketAddress(host, port), TIMEOUT);
            return true;
        } catch (Exception e){
            return false;
        } finally {
            if (socket!=null){
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

    对于实际的service服务,在设计上没有考虑线程安全问题,一方面是因为该服务检测相关的api的访问都是管理员身份鉴权通过才能被允许的,正常的操作流程是不会出现安全问题的,另一方面,嗯,因为懒(误打QAQ)。

DsapServiceImpl 类 :DetectedServerAndPort的缩写,提供了api所需要的6个服务实现

  • 只要是增删改类型的api,在返回前都进行一轮状态更新 updateAll()。
  • selectAll方法有两个,一个是private访问权限的 selectAll0(),实际的进行了redis数据库的查询,并且将处理后的数据返回,而提供给api访问的public访问权限的selectAll(),是先从redis缓存中取直接结果,没有取到的的话说明是第一次访问,走一遍updateAll后再一次从redis取结果就一定存在了。
  • addNewPort 方法,因为设计上的value是一个json字符串,无法直接插入一个端口,所以必要的先从redis中取出该ip的端口信息json字符串。然而设置的序列化方式是 Jackson2JsonRedisSerializer ,ops.get()得到的数据类型实际是 ArrayList<LinkedHashMap> ,通过类型修正得到真正的List<DetectedPort>类型集合(后文详细解释这里的处理流程),因为要保证端口的唯一性,而结果list是无法去重的,所以用list.removeIf方法先把已经存在的该port数据删除,然后再把当前新的数据加入进来。最后通过ops.put把该ip的value替换成新的结果。
  • addServer 方法,相比于addNewPort方法就比较简单了,如果发现添加了相同的,直接替换即可。
@Service
public class DsapServiceImpl implements DsapService {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private AsyncService asyncService;

    private static final String DSAP_PORTS = "DSAP-PORT";
    private static final String DSAP_SERVER = "DSAP-SERVER";
    /**
     * dsap状态缓存
     */
    private static final String DSAP_CACHE = "DSAP-CACHE";

    /**
     * 添加一个端口
     * @param dp 端口信息
     * @param ip ip
     * @return 1:up;-1:down
     */
    @Override
    public RetResult<Integer> addNewPort(DetectedPort dp, String ip) {
        boolean pd = DsapUtil.portDetect(ip, dp.getPort());
        dp.setUp(pd);

        HashOperations<String, String, List<DetectedPort>> ops = redisTemplate.opsForHash();

        // 获取当前ip的端口列表
        List<DetectedPort> ports = ops.get(DSAP_PORTS, ip);

        // 添加当前新的端口到列表
        if (ports == null) ports = new ArrayList<>();
        ports = transfer(ports);
        ports.removeIf(port -> port.getPort().equals(dp.getPort()));
        ports.add(dp);

        // 设置到hash中
        ops.put(DSAP_PORTS, ip, ports);

        // 更新服务器装填
        HashOperations<String, String, Boolean> ops2 = redisTemplate.opsForHash();
        boolean sd = DsapUtil.serverDetect(ip);
        ops2.put(DSAP_SERVER, ip, sd);

        updateAll();
        return RetResult.success(pd?1:-1);
    }

    /**
     * 删除指定的端口,并且返回缓存的状态
     * @param ip ip
     * @param port port
     * @return ret
     */
    @Override
    public RetResult<DetectedServer> delPort(String ip, int port) {
        // 获取server装填
        HashOperations<String,String,Boolean> ops1 = redisTemplate.opsForHash();
        Boolean up = ops1.get(DSAP_SERVER, ip);
        if (up == null) return RetResult.fail(ip + "不存在的记录");

        HashOperations<String, String, List<DetectedPort>> ops = redisTemplate.opsForHash();
        List<DetectedPort> ports = ops.get(DSAP_PORTS, ip);

        // 删除指定监听的端口
        if (ports == null) ports = new ArrayList<>();
        ports = transfer(ports);
        ports.removeIf(_port -> _port.getPort() == port);
        ops.put(DSAP_PORTS, ip, ports);

        updateAll();
        return RetResult.success(new DetectedServer(ip, up, ports));
    }

    /**
     * 添加一个服务器
     * @param ip ip
     * @return 1:up; -1:down
     */
    @Override
    public RetResult<Integer> addServer(String ip) {
        HashOperations<String,String,Boolean> ops1 = redisTemplate.opsForHash();
        boolean up = DsapUtil.serverDetect(ip);
        ops1.put(DSAP_SERVER, ip, up);

        updateAll();
        return RetResult.success(up?1:-1);
    }

    @Override
    public RetResult<Integer> delServer(String ip) {
        HashOperations ops = redisTemplate.opsForHash();
        ops.delete(DSAP_SERVER, ip);
        ops.delete(DSAP_PORTS, ip);
        updateAll();
        return RetResult.success(1);
    }

    /**
     * 从缓存获取所有服务状态
     * @return ret
     */
    public String selectAll() {
        ValueOperations<String, String> cache = redisTemplate.opsForValue();
        String res = cache.get(DSAP_CACHE);
        if (res != null) return res;
        updateAll();
        return cache.get(DSAP_CACHE);
    }

    /**
     * 提供给 {@link #updateAll()} 使用的内部select方法
     * @return 结果
     */
    private List<DetectedServer> selectAll0() {
        HashOperations<String, String, List<DetectedPort>> ops = redisTemplate.opsForHash();
        Map<String, List<DetectedPort>> ports = ops.entries(DSAP_PORTS);

        HashOperations<String,String,Boolean> ops1 = redisTemplate.opsForHash();
        Map<String, Boolean> entries = ops1.entries(DSAP_SERVER);
        Iterator<Map.Entry<String, Boolean>> iterator = entries.entrySet().iterator();
        List<DetectedServer> servers = new ArrayList<>(entries.size());
        while (iterator.hasNext()){
            Map.Entry<String, Boolean> next = iterator.next();
            String ip = next.getKey();
            List<DetectedPort> tmp = ports.get(ip);
            if (tmp == null) tmp = new ArrayList<>();
            servers.add(new DetectedServer(ip,next.getValue(), tmp));
        }

        return servers;
    }

    /**
     * 定时任务定时的更新状态, 并且将结果写入cache, cache的结果是一个ret
     */
    @Override
    public void updateAll() {
        BoundHashOperations<String, String, Boolean> ops1 = redisTemplate.boundHashOps(DSAP_SERVER);
        BoundHashOperations<String, String, List<DetectedPort>> ops = redisTemplate.boundHashOps(DSAP_PORTS);
        ValueOperations<String, String> cache = redisTemplate.opsForValue();

        List<DetectedServer> data = selectAll0();
        Map<String, Boolean> serverStatus = new HashMap<>(data.size());
        Map<String, List<DetectedPort>> portsStatus = new HashMap<>();
        StringBuilder msg = new StringBuilder();
        for (DetectedServer server:data) {
            String ip = server.getIp();

            // 检测当前服务器up
            boolean serverUp = DsapUtil.serverDetect(ip);
            serverStatus.put(ip, serverUp);
            server.setUp(serverUp);

            // 检测每个端口的状态
            List<DetectedPort> tmp = new ArrayList<>(server.getPorts().size());
            server.setPorts(transfer(server.getPorts()));
            for (DetectedPort port : server.getPorts()) {
                boolean portUp = DsapUtil.portDetect(ip, port.getPort());
                port.setUp(portUp);
                // 如果服务下线并且需要通知, 那么进行一次通知, 并且关闭通知, 避免密集通知
                if (!portUp && port.getAlert()) {
                    // <b>ip:port name<b/><br/>
                    msg.append("<b>").append(ip).append(":").append(port.getPort()).append(" ").append(port.getName()).append("</b></br>");
                    port.setAlert(false);
                }
                tmp.add(port);
            }
            portsStatus.put(ip, tmp);
        }

        // 进行一批次的通知,异步发送邮件
        if (!"".equals(msg.toString())){
            asyncService.serviceDownAlert(msg.toString());
        }

        // 更新redis中数据的状态
        ops1.putAll(serverStatus);
        ops.putAll(portsStatus);
        // 将数据写入cache中
        JSONObject jsonObject = new JSONObject(RetResult.success(data));
        cache.set(DSAP_CACHE, jsonObject.toString(), 10, TimeUnit.MINUTES);
    }

    /**
     * 将redis的get得到的原始数据修正成正确格式的对象
     * @param raw 从redis的get出来的原始数据
     * @return 正确的格式数据
     */
    private List<DetectedPort> transfer(List<DetectedPort> raw){
        String tmp = JSONObject.valueToString(raw);
        Type type = new TypeToken<List<DetectedPort>>() { }.getType();
        return new Gson().fromJson(tmp, type);
    }
}

服务监测结果展示

首先看一下前端后台的效果图,可以发现加入了四台服务器,每个服务器有相应的端口监测,“√”/“x”表示是否开启该端口异常下线的邮件通知。

接着看一下redis数据库的结果:

首先是这次服务监测产生的三个key都正常存在了,然后其中DSAP-CACHE缓存的结果也正常(不过看着好多转义符号啊2333).


查看DSAP-SERVER与DSAP-PORT的kv,也都正常。

redis序列化

序列化器:

    默认的,redis对对象的序列化方式是 JdkSerializationRedisSerializer ,这种方式的结果因为是二进制数据,不借用专门的功能难以查阅结果,虽然反序列化时候不需要类型信息,但是缺点也很明显:被序列化的类需要实现Serializable接口、结果占用空间比较大。

    StringRedisSerializer,只适合字符串类型数据的序列化,通常作为各种key的序列化方式。

    GenericToStringSerializer,比上面一个更加通用,可以传递类型后,生成字符串。

    Jackson2JsonRedisSerializer,比较常用的一种对值的序列化方式,结果非常简洁,速度快,但是需要一个序列化对象的类型作为参数。

    GenericJackson2JsonRedisSerializer,比上面一个更加通用,存储的结果中会保存类型信息,所以结果占用会大一些。

    当然还有其它的序列化器,比如阿里的FastJson2JsonRedisSerializer,这里就不做过多介绍了。

redisTemplate:

    对于springboot的redis-starter自动配置类RedisAutoConfiguration来说,如果我们没有配置redisTemplate命名的RedisTemplate对象或者stringRedisTemplate类型的对象,那么也都会自动注入一个自动配置的对象。

    当然自动配置的redisTemplate并没有设置序列化器,所以我们可以有两种方式来更改,一种是将默认的引入到参数列表中然后直接修改,实际的对象只在堆中存在一个,也可以起redisTemplate名称的函数直接注册,这样就自动配置类的注册就不会生效了。我下面是第一种实现方式:

  @Bean
    public RedisTemplate<Object, Object> redisStringTemplate(
            RedisTemplate<Object, Object> redisTemplate, ServerProperties properties){
        Jackson2JsonRedisSerializer<?> jksSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
        StringRedisSerializer strSerializer = new StringRedisSerializer();

        redisTemplate.setKeySerializer(jksSerializer);
        redisTemplate.setValueSerializer(jksSerializer);

        redisTemplate.setHashKeySerializer(strSerializer);
        redisTemplate.setHashValueSerializer(jksSerializer);

        initData(redisTemplate, properties);
        return redisTemplate;
    }

    可以看到我这里对hash的值使用了Jackson2JsonRedisSerializer,并且传入的序列化类型是Object.class,也就是最通用的。这也就造成了一个问题,那就是如果value是带泛型的数据,那么最后无法解析到精确到泛型特定类型的数据,例如List<XXX>,最终只能解析成ArrayList<LinkedHashMap>类型的数据,LinkedHashMap用来组装具体的数据kv(属性名、属性值),而非我们特定的XXX类型。具体的原理可以参考这两篇文章:

https://blog.csdn.net/qq_38074398/article/details/128233005 (从结论上解析)【1】

https://www.cnblogs.com/kzyuan/p/16312931.html (从底层设计细节解析)【2】

GenericJackson2JsonRedisSerializer根据额外插入的类全限定名通过反射可以正确得到实体类的实例。

而Jackson2JsonRedisSerializer由于没有插入额外的信息,那么只能通过不同的数据结构来组装反序列化后的内容。对于List的数据会反序列化成ArrayList<LinkedHashMap>,使用LinkedHashMap来组装实体类对象的字段与字段值。

泛型的解析:

    那么如何讲ArrayList<LinkedHashMap>转为特定的类型呢?一种解决方式是上方【2】链接文章中提到的,使用ObjectMapper进行类型转化。

    @Test
    public void testRange() {
        String key = "right_push_all_01";
        List<LinkedHashMap<String, Object>> linkedHashMapList = redisService.lRange(key, 0, -1);
        ObjectMapper objectMapper = new ObjectMapper();
        List<ThisIsDTO> thisIsDTOList = objectMapper.convertValue(linkedHashMapList, new TypeReference<List<ThisIsDTO>>() { });
        for (ThisIsDTO thisIsDTO : thisIsDTOList) {
            System.out.println(thisIsDTO.getAge());
        }
    }

    当然我这里采用另外一种取巧的方式,就是先转为json字符串,再使用gson进行类型修正,当然也可以不返回,直接raw = new Gson...也可以。

    private List<DetectedPort> transfer(List<DetectedPort> raw){
        String tmp = JSONObject.valueToString(raw);
        Type type = new TypeToken<List<DetectedPort>>() { }.getType();
        return new Gson().fromJson(tmp, type);
    }

    这样就可以正确的当作特定类型的List来操作了。


上一篇:win10的一些问题解决
下一篇:Springboot的jar包分离