淘宝Tedis组件究竟是个啥(一)

淘宝的Tedis组件究竟是个啥呢?可能有一些朋友没有听过这个名字,有一些朋友会经常使用,那么今天我就来和大家深入分析一下,它的使用和原理。

一、Tedis简介

Tedis是另一个redis的java客户端,Tedis的目标是打造一个可在生产环境直接使用的高可用Redis解决方案。

特性如下:

  • 高可用:Tedis使用多写随机读做HA确保redis的高可用
  • 高性能:使用特殊的线程模型,使redis的性能不限制在客户端
  • 多种使用方式:如果你只有一个redis实例,并不需要tedis的HA功能,可以直接使用tedis-atomic;使用tedis的高可用功能需要部署多个redis实例使用tedis-group
  • 两种API:包括针对byte的底层api和面向object的高层api
  • 多种方便使用redis的工具集合:包括mysql数据同步到redis工具,利用redis做搜索工具等

二、Tedis能做啥

1、Tedis的原理 Tedis是对开源的Redis客户端组件Jedis进行的封装,在Jedis的基础上封装了一套更易于使用的byte api和object api接口,在部署上使用的是master-master结构,实现了多写与随机读的机制。既:每个写请求都会发到所有服务器上面,每个读请求随机读取一个服务器,当在某个服务器读取失败后,将该台服务器加到重试队列中,直到服务器恢复正常客户端请求才会重新访问到该服务器。

2、典型的应用场景

Paste_Image.png

本图选自周成的《聚划算架构演化与系统优化》

二、Tedis的简单使用

可以先在Maven中依赖下面的jar包,当然还有更高版本的。

<dependency> 
    <groupId>com.taobao.common</groupId>
    <artifactId>tedis-group</artifactId>
    <version>1.1.0</version>
</dependency>

最简单的使用方式就是这样,只需要声明一个TedisGroup类并且初始化,然后就可以用ValueCommands进行set,get等操作了。唯一有点别扭的可能就是Tedis自己封装了一些命令类,需要我们自己去定义然后使用。

Group tedisGroup = new TedisGroup(appName, version);
tedisGroup.init();
ValueCommands valueCommands = new DefaultValueCommands(tedisGroup.getTedis());
// 写入一条数据
valueCommands.set(1, "test", "test value object");
// 读取一条数据
valueCommands.get(1, "test");

三、Tedis的源码分析

1、Tedis的包结构

Paste_Image.png

注:由于可以看出Tedis的结构实际分为了四个工程: tedis-atomic、tedis-common、tedis-group、tedis-replicator、tedis-search。

  • tedis-atomic

Paste_Image.png

注:这个工程主要是集中一些Tedis的初始化类,像客户端类,连接类,初始化工厂类等等。

  • tedis-common

Paste_Image.png

注:这个工程主要就是集中了整个Tedis的工具类,异常类,线程类和监控类等。

  • tedis-group

Paste_Image.png

注:这个工程也是我们在源码分析中重点介绍的,里面汇集了Tedis最核心的代码,像集群,多写随机读等功能都在此实现。

  • tedis-relicator

Paste_Image.png

注:不是本文重点,不做解释,感兴趣的读者可以自己了解。

  • tedis-search

Paste_Image.png

注:不是本文重点,不做解释,感兴趣的读者可以自己了解。

2、tedis-group源码解析 我们看源码往往是通过使用的类开始的,也就是上面提到的TedisGroup类开始,

  • 我们先来看TedisGroup类结构图:

Paste_Image.png

注:TedisGroup类是一个非常核心的类,有一个很重要的方法就是init和一个很重要的内部类RedisGroupInvocationHandler,在TedisGroup中依赖了二个重要的类,ConfigManager和RedisCommands。

  • ConfigManager解析

Paste_Image.png 注:这个类的作用是: A、通过从Zookeeper、File、Diamond中解析有哪些Redis服务器,一共有几台服务器,超时时间和失败策略是如何。 B、写入多个Redis和读取Redis的策略是怎么样的,在这个类中是用的Router路由的方式。

  • RedisCommands解析

Paste_Image.png

注:Tedis组件把Redis的操作命令全部封装到RedisCommands接口中,最底层的实现类是Tedis,通过这个类可以实现多个Redis的写和随机读等策略,Tedis类的部分源码如下:

    @Override
    public List<byte[]> sort(byte[] key, SortParams params) {
        try {
            if (params == null) {
                client.sort(key);
            } else {
                client.sort(key, params);
            }
            return client.getBinaryMultiBulkReply();
        } catch (Exception ex) {
            throw new TedisException(ex);
        }
    }


    @Override
    public Boolean ping() {
        client.ping();
        return PONG.equals(client.getStatusCodeReply());
    }

    @Override
    public Long del(byte[]... keys) {
        client.del(keys);
        return client.getIntegerReply();
    }

    @Override
    public Boolean exists(byte[] key) {
        client.exists(key);
        return client.getIntegerReply() == 1;
    }

    @Override
    public Set<byte[]> keys(byte[] pattern) {
        client.keys(pattern);
        return new HashSet<byte[]>(client.getBinaryMultiBulkReply());
    }

    @Override
    public Boolean persist(byte[] key) {
        client.persist(key);
        return client.getIntegerReply() == 1;

    }

    @Override
    public Boolean move(byte[] key, int dbIndex) {
        client.move(key, dbIndex);
        return client.getIntegerReply() == 1;
    }

    @Override
    public byte[] randomKey() {
        client.randomKey();
        return client.getBinaryBulkReply();
    }

     @Override
    public byte[] get(byte[] key) {
        client.get(key);
        return client.getBinaryBulkReply();
    }

    @Override
    public Boolean set(byte[] key, byte[] value) {
        client.set(key, value);
        return OK.equals(client.getStatusCodeReply());
    }
  • RedisGroupInvocationHandler内部类解析

这个类使用的是动态代理的模式,实现的是InvocationHandler接口,核心方法依然是 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable 另外还有一个关键的注解就是Process,在RedisCommands的实现类Tedis中有使用,比如:

   @Process(Policy.WRITE)
    public Long zAdd(@ShardKey byte[] key, Tuple... value) {
        client.zadd(key, value);
        return client.getIntegerReply();
    }

标识了这个注解的方法在代理中会进行判断是写还是读操作,如果是写操作则会读取所有Redis配置以循环的方式逐一插入数据,如果其中一个Redis报错则记录日志抛出异常,如果是读操作则采用RandomRouter的方式随机从Redis列表中选取一个进行读取操作。具体部分源码实现如下:

       @Override
        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
            long time = System.currentTimeMillis();
            String name = method.getName();
            Router rr = cm.getRouter();
            Process annotation = method.getAnnotation(Process.class);
            Throwable exception = null;
            if (annotation.value() == Policy.READ) {
                while (rr.getRouteData().props.size() > 0) {
                    Single s = rr.route();
                    try {
                        Object result = method.invoke(s.getTedis(), args);
                        statLog(name, true, time);
                        return result;
                    } catch (Throwable t) {
                        statLog(name, false, time);
                        exception = t;
                        logger.warn("read exception:" + s.getProperties(), t);
                        boolean connectionError = false;
                        try {
                            if (t instanceof InvocationTargetException) {// 解包异常
                                InvocationTargetException ite = (InvocationTargetException) t;
                                UndeclaredThrowableException ute = (UndeclaredThrowableException) ite.getTargetException();
                                if (ute.getUndeclaredThrowable() instanceof TimeoutException) {
                                    connectionError = true;
                                    rr.onError(s);
                                } else {
                                    ExecutionException ee = (ExecutionException) ute.getUndeclaredThrowable();
                                    InvocationTargetException ite_1 = (InvocationTargetException) ee.getCause();
                                    TedisException te = (TedisException) ite_1.getTargetException();
                                    if (te.getCause() instanceof TedisConnectionException) {
                                        connectionError = true;
                                        rr.onError(s);
                                    }
                                }
                            }
                        } catch (Throwable tt) {
                            logger.warn("解包异常:", tt);
                            // 可能会抛出转换异常,符合预期,如果碰到转换异常,直接在connection error
                            // 过程中从新抛出
                        }

                        if (!connectionError) {
                            throw t;
                        }
                    }
                }

                throw new Exception("read RouteData is empty," + rr, exception);
            } else if (annotation.value() == Policy.WRITE) {
                Single[] ss = rr.getRouteData().group;
                if (ss == null || ss.length == 0) {
                    throw new Exception("write RouteData is empty," + rr, exception);
                }
                Object result = null;
                int e = 0;
                for (Single s : ss) {
                    try {
                        result = method.invoke(s.getTedis(), args);
                    } catch (Throwable t) {
                        e++;
                        statLog(name, false, time);
                        logger.warn("write exception:" + s.getProperties(), t);
                        exception = t;
                        try {
                            // 解包异常
                            InvocationTargetException ite = (InvocationTargetException) t;
                            UndeclaredThrowableException ute = (UndeclaredThrowableException) ite.getTargetException();
                            if (ute.getUndeclaredThrowable() instanceof TimeoutException) {
                                rr.onError(s);
                            } else {
                                ExecutionException ee = (ExecutionException) ute.getUndeclaredThrowable();
                                InvocationTargetException ite_1 = (InvocationTargetException) ee.getCause();
                                TedisException te = (TedisException) ite_1.getTargetException();
                                if (te.getCause() instanceof TedisConnectionException) {
                                    rr.onError(s);
                                }
                            }
                        } catch (Throwable tt) {
                            logger.warn("解包异常:", tt);
                        }
                    }
                }

                if (e >= 2) {// 全部都抛异常了,告知调用端
                    throw exception;
                }
                statLog(name, true, time);
                return result;
            } else if ("toString".equals(name)) {
                return "";
            } else if ("hashCode".equals(name)) {
                Single s = rr.route();
                if (s != null) {
                    return s.hashCode();
                } else {
                    return 0;
                }
            } else if ("equals".equals(name)) {
                Single s = rr.route();
                if (args.length == 1) {
                    return s.equals(args[0]);
                }
            }
            statLog(name, false, time);
            throw new Exception("method don't match:" + name);
        }
    }

对于整体架构和其他源码的分析,我们将在第二篇进行深入分析,请大家关注。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏FreeBuf

缓冲区溢出攻击初学者手册(更新版)

说明 之前版本翻译质量不佳,本人赵阳在这里对本文的读者表示深深的歉意。由于本人的疏忽和大意导致您不能很好的读完这篇文章,同时也对原文内容进行了破坏,也对IDF和...

2819
来自专栏安恒网络空间安全讲武堂

从零基础到成功解题之0ctf-ezdoor

2144
来自专栏微信公众号:Java团长

从并发编程到分布式系统——如何处理海量数据(上)

在这里想写写自己在学习并发处理的学习思路,也会聊聊自己遇到的那些坑,以此为记,希望鞭策自己不断学习、永不放弃!

771
来自专栏大史住在大前端

webpack4.0各个击破(5)—— Module篇

使用webpack对脚本进行合并是非常方便的,因为webpack实现了对各种不同模块规范的兼容处理,对前端开发者来说,理解这种实现方式比学习如何配置webpac...

1282
来自专栏与神兽党一起成长

解析XML和JSON内容的一点技巧

在没有统一标准的情况下,一个系统对接多个外部系统往往会遇到请求接口响应数据异构的情况,有可能返回的是XML,也有可能返回 JSON。除了返回类型不同,内容结构也...

1992
来自专栏FD的专栏

一步步理解python的异步IO

看到越来越多的大佬都在使用python的异步IO,协程等概念来实现高效的IO处理过程,可是我对这些概念还不太懂,就学习了一下。 因为是初学者,在理解上有很多不到...

1562
来自专栏java思维导图

Java中高级面试题

技术文章第一时间送达! 本文作者是CyanQueen,欢迎点击阅读原文 一.基础知识: 1)集合类:List和Set比较,各自的子类比较(ArrayList,V...

3725
来自专栏java一日一条

2017年高频率的互联网校园招聘面试题

参加了2017年校招,面试了阿里、百度、腾讯、滴滴、美团、网易、去哪儿等公司,个人是客户端 Android 方向,总结了面试过程中频率出现较高的题目,希望对大家...

1742
来自专栏码云1024

MFC多线程

4476
来自专栏君赏技术博客

建议大型项目用上Try Catch建议大型项目用上Try Catch

我们在平时项目做功能的时候,经常会遇到崩溃的情况。如果是我们在开发测试阶段,我们可以找到原因修复。但是遇到已经上线,出现这种问题。要么使用JSPatch进行热修...

801

扫码关注云+社区

领取腾讯云代金券