2024-01-18  阅读(47)
原文作者:hashcon 原文地址: https://zhanghaoxin.blog.csdn.net/article/details/51301799

4.配置模块

每个MyCatServer初始化时,会初始化:
MyCatServer.java:

    public static final String NAME = "MyCat";
    private static final long LOG_WATCH_DELAY = 60000L;
    private static final long TIME_UPDATE_PERIOD = 20L;
    private static final MycatServer INSTANCE = new MycatServer();
    private static final Logger LOGGER = LoggerFactory.getLogger("MycatServer");
    private final RouteService routerService;
    private final CacheService cacheService;
    private Properties dnIndexProperties;
    //AIO连接群组
    private AsynchronousChannelGroup[] asyncChannelGroups;
    private volatile int channelIndex = 0;
    
    //全局序列号
    private final MyCATSequnceProcessor sequnceProcessor = new MyCATSequnceProcessor();
    private final DynaClassLoader catletClassLoader;
    private final SQLInterceptor sqlInterceptor;
    private volatile int nextProcessor;
    private BufferPool bufferPool;
    private boolean aio = false;
    
    //XA事务全局ID生成
    private final AtomicLong xaIDInc = new AtomicLong();
    private MycatServer() {
        //读取文件配置
        this.config = new MycatConfig();
        //定时线程池,单线程线程池
        scheduler = Executors.newSingleThreadScheduledExecutor();
        //SQL记录器
        this.sqlRecorder = new SQLRecorder(config.getSystem()
                .getSqlRecordCount());
        /**
         * 是否在线,MyCat manager中有命令控制
         * | offline | Change MyCat status to OFF |
         * | online | Change MyCat status to ON |
         */
        this.isOnline = new AtomicBoolean(true);
        //缓存服务初始化
        cacheService = new CacheService();
        //路由计算初始化
        routerService = new RouteService(cacheService);
        // load datanode active index from properties
        dnIndexProperties = loadDnIndexProps();
        try {
            //SQL解析器
            sqlInterceptor = (SQLInterceptor) Class.forName(
                    config.getSystem().getSqlInterceptor()).newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        //catlet加载器
        catletClassLoader = new DynaClassLoader(SystemConfig.getHomePath()
                + File.separator + "catlet", config.getSystem()
                .getCatletClassCheckSeconds());
        //记录启动时间
        this.startupTime = TimeUtil.currentTimeMillis();
    }

第一步是读取文件配置,主要是三个文件:schema.xml,rule.xml和server.xml. 读取后的配置会加载到MyCatConfig中。
MyCatConfig.java:

    public MycatConfig() {
    //读取schema.xml,rule.xml和server.xml
    ConfigInitializer confInit = new ConfigInitializer(true);
    this.system = confInit.getSystem();
    this.users = confInit.getUsers();
    this.schemas = confInit.getSchemas();
    this.dataHosts = confInit.getDataHosts();
    
    this.dataNodes = confInit.getDataNodes();
    for (PhysicalDBPool dbPool : dataHosts.values()) {
        dbPool.setSchemas(getDataNodeSchemasOfDataHost(dbPool.getHostName()));
    }
    this.quarantine = confInit.getQuarantine();
    this.cluster = confInit.getCluster();
    //初始化重加载配置时间
    this.reloadTime = TimeUtil.currentTimeMillis();
    this.rollbackTime = -1L;
    this.status = RELOAD;
    //配置加载锁
    this.lock = new ReentrantLock();
    }

它们都通过ConfigInitializer读取:

    public ConfigInitializer(boolean loadDataHost) {
        //读取schema.xml
        SchemaLoader schemaLoader = new XMLSchemaLoader();
        //读取server.xml
        XMLConfigLoader configLoader = new XMLConfigLoader(schemaLoader);
        schemaLoader = null;
        //加载配置
        this.system = configLoader.getSystemConfig();
        this.users = configLoader.getUserConfigs();
        this.schemas = configLoader.getSchemaConfigs();
        //是否重新加载DataHost和对应的DataNode
        if (loadDataHost) {
            this.dataHosts = initDataHosts(configLoader);
            this.dataNodes = initDataNodes(configLoader);
        }
        //权限管理
        this.quarantine = configLoader.getQuarantineConfig();
        this.cluster = initCobarCluster(configLoader);
        //不同类型的全局序列处理器的配置加载
        if (system.getSequnceHandlerType() == SystemConfig.SEQUENCEHANDLER_MYSQLDB) {
            IncrSequenceMySQLHandler.getInstance().load();
        }
        if (system.getSequnceHandlerType() == SystemConfig.SEQUENCEHANDLER_LOCAL_TIME) {
            IncrSequenceTimeHandler.getInstance().load();
        }
        //检查user与schema配置对应以及schema配置不为空
        this.checkConfig();
    }

4.1 rule.xml

读取schema之前会先读取rule.xml。
XmlSchemaLoader.java:

    public XMLSchemaLoader(String schemaFile, String ruleFile) {
        //先读取rule.xml
        XMLRuleLoader ruleLoader = new XMLRuleLoader(ruleFile);
        this.tableRules = ruleLoader.getTableRules();
        ruleLoader = null;
        this.dataHosts = new HashMap<String, DataHostConfig>();
        this.dataNodes = new HashMap<String, DataNodeConfig>();
        this.schemas = new HashMap<String, SchemaConfig>();
        //读取加载schema配置
        this.load(DEFAULT_DTD, schemaFile == null ? DEFAULT_XML : schemaFile);
    }
    
    public XMLSchemaLoader() {
        this(null, null);
    }

XMLRuleLoader.java:

    public XMLRuleLoader(String ruleFile) {
        // this.rules = new HashSet<RuleConfig>();
        //rule名 -> rule
        this.tableRules = new HashMap<String, TableRuleConfig>();
        //function名 -> 具体分片算法
        this.functions = new HashMap<String, AbstractPartitionAlgorithm>();
        //默认为:/rule.dtd和/rule.xml
        load(DEFAULT_DTD, ruleFile == null ? DEFAULT_XML : ruleFile);
    }
    
    public XMLRuleLoader() {
        this(null);
    }
    private void load(String dtdFile, String xmlFile) {
        InputStream dtd = null;
        InputStream xml = null;
        try {
            dtd = XMLRuleLoader.class.getResourceAsStream(dtdFile);
            xml = XMLRuleLoader.class.getResourceAsStream(xmlFile);
            //读取出语意树
            Element root = ConfigUtil.getDocument(dtd, xml)
                    .getDocumentElement();
            //加载Function
            loadFunctions(root);
            //加载TableRule
            loadTableRules(root);
        } catch (ConfigException e) {
            throw e;
        } catch (Exception e) {
            throw new ConfigException(e);
        } finally {
            if (dtd != null) {
                try {
                    dtd.close();
                } catch (IOException e) {
                }
            }
            if (xml != null) {
                try {
                    xml.close();
                } catch (IOException e) {
                }
            }
        }
    }

ConfigUtil.java解析语意树:

    public static Document getDocument(final InputStream dtd, InputStream xml) throws ParserConfigurationException,
                SAXException, IOException {
        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
        factory.setValidating(true);
        factory.setNamespaceAware(false);
        DocumentBuilder builder = factory.newDocumentBuilder();
        builder.setEntityResolver(new EntityResolver() {
            @Override
            public InputSource resolveEntity(String publicId, String systemId) {
                return new InputSource(dtd);
            }
        });
        builder.setErrorHandler(new ErrorHandler() {
            @Override
            public void warning(SAXParseException e) {
            }
    
            @Override
            public void error(SAXParseException e) throws SAXException {
                throw e;
            }
    
            @Override
            public void fatalError(SAXParseException e) throws SAXException {
                throw e;
            }
        });
        return builder.parse(xml);
    }

加载functions, XmlRuleLoader.java

    private void loadFunctions(Element root) throws ClassNotFoundException,
                InstantiationException, IllegalAccessException,
                InvocationTargetException {
            NodeList list = root.getElementsByTagName("function");
            for (int i = 0, n = list.getLength(); i < n; ++i) {
                Node node = list.item(i);
                if (node instanceof Element) {
                    Element e = (Element) node;
                    //获取name标签
                    String name = e.getAttribute("name");
                    //如果Map已有,则function重复
                    if (functions.containsKey(name)) {
                        throw new ConfigException("rule function " + name
                                + " duplicated!");
                    }
                    //获取class标签
                    String clazz = e.getAttribute("class");
                    //根据class利用反射新建分片算法
                    AbstractPartitionAlgorithm function = createFunction(name, clazz);
    
                    ParameterMapping.mapping(function, ConfigUtil.loadElements(e));
                    //每个AbstractPartitionAlgorithm可能会实现init来初始化
                    function.init();
                    //放入functions map
                    functions.put(name, function);
                }
            }
        }
    
    private AbstractPartitionAlgorithm createFunction(String name, String clazz)
            throws ClassNotFoundException, InstantiationException,
            IllegalAccessException, InvocationTargetException {
        Class<?> clz = Class.forName(clazz);
        //判断是否继承AbstractPartitionAlgorithm
        if (!AbstractPartitionAlgorithm.class.isAssignableFrom(clz)) {
            throw new IllegalArgumentException("rule function must implements "
                    + AbstractPartitionAlgorithm.class.getName() + ", name=" + name);
        }
        return (AbstractPartitionAlgorithm) clz.newInstance();
    }

加载所有的function的node,每一个node就是一个AbstractPartitionAlgorithm,并放入functions这个map中;

    private final Map<String, TableRuleConfig> tableRules;

对于每一个node,通过反射新建对应参数的AbstractPartitionAlgorithm。这样,所有的function就加载到了functions这个map中。
同理,加载TableRule,就加上了function是否存在的判断:

    /**
    * tableRule标签结构:
     * <tableRule name="sharding-by-month">
     *     <rule>
     *         <columns>create_date</columns>
     *         <algorithm>partbymonth</algorithm>
     *     </rule>
     * </tableRule>
     * @param root
     * @throws SQLSyntaxErrorException
        */
    private void loadTableRules(Element root) throws SQLSyntaxErrorException {
        //获取每个tableRule标签
        NodeList list = root.getElementsByTagName("tableRule");
        for (int i = 0, n = list.getLength(); i < n; ++i) {
            Node node = list.item(i);
            if (node instanceof Element) {
                Element e = (Element) node;
                //先判断是否重复
                String name = e.getAttribute("name");
                if (tableRules.containsKey(name)) {
                    throw new ConfigException("table rule " + name
                            + " duplicated!");
                }
                //获取rule标签
                NodeList ruleNodes = e.getElementsByTagName("rule");
                int length = ruleNodes.getLength();
                if (length > 1) {
                    throw new ConfigException("only one rule can defined :"
                            + name);
                }
                //目前只处理第一个,未来可能有多列复合逻辑需求
                //RuleConfig是保存着rule与function对应关系的对象
                RuleConfig rule = loadRule((Element) ruleNodes.item(0));
                String funName = rule.getFunctionName();
                //判断function是否存在,获取function
                AbstractPartitionAlgorithm func = functions.get(funName);
                if (func == null) {
                    throw new ConfigException("can't find function of name :"
                            + funName);
                }
                rule.setRuleAlgorithm(func);
                //保存到tableRules
                tableRules.put(name, new TableRuleConfig(name, rule));
            }
        }
    }

这样,所有的tableRule和function就加载完毕。保存在一个变量中,就是tableRules:
XMLRuleLoader.java:

    private final Map<String, TableRuleConfig> tableRules;

4.2 schema.xml:

    public XMLSchemaLoader(String schemaFile, String ruleFile) {
        //先读取rule.xml
        XMLRuleLoader ruleLoader = new XMLRuleLoader(ruleFile);
        //将tableRules拿出,用于这里加载Schema做rule有效判断,以及之后的分片路由计算
        this.tableRules = ruleLoader.getTableRules();
        //释放ruleLoader
        ruleLoader = null;
        this.dataHosts = new HashMap<String, DataHostConfig>();
        this.dataNodes = new HashMap<String, DataNodeConfig>();
        this.schemas = new HashMap<String, SchemaConfig>();
        //读取加载schema配置
        this.load(DEFAULT_DTD, schemaFile == null ? DEFAULT_XML : schemaFile);
    }
    
    private void load(String dtdFile, String xmlFile) {
        InputStream dtd = null;
        InputStream xml = null;
        try {
            dtd = XMLSchemaLoader.class.getResourceAsStream(dtdFile);
            xml = XMLSchemaLoader.class.getResourceAsStream(xmlFile);
            Element root = ConfigUtil.getDocument(dtd, xml).getDocumentElement();
            //先加载所有的DataHost
            loadDataHosts(root);
            //再加载所有的DataNode
            loadDataNodes(root);
            //最后加载所有的Schema
            loadSchemas(root);
        } catch (ConfigException e) {
            throw e;
        } catch (Exception e) {
            throw new ConfigException(e);
        } finally {
    
            if (dtd != null) {
                try {
                    dtd.close();
                } catch (IOException e) {
                }
            }
    
            if (xml != null) {
                try {
                    xml.close();
                } catch (IOException e) {
                }
            }
        }
    }

先看下DataHostConfig这个类的结构:

202401182020355331.png
XMLSchemaLoader.java:

    private void loadDataHosts(Element root) {
        NodeList list = root.getElementsByTagName("dataHost");
        for (int i = 0, n = list.getLength(); i < n; ++i) {
    
            Element element = (Element) list.item(i);
            String name = element.getAttribute("name");
            //判断是否重复
            if (dataHosts.containsKey(name)) {
                throw new ConfigException("dataHost name " + name + "duplicated!");
            }
            //读取最大连接数
            int maxCon = Integer.valueOf(element.getAttribute("maxCon"));
            //读取最小连接数
            int minCon = Integer.valueOf(element.getAttribute("minCon"));
            /**
             * 读取负载均衡配置
             * 1. balance="0", 不开启分离机制,所有读操作都发送到当前可用的 writeHost 上。
             * 2. balance="1",全部的 readHost 和 stand by writeHost 参不 select 的负载均衡
             * 3. balance="2",所有读操作都随机的在 writeHost、readhost 上分发。
             * 4. balance="3",所有读请求随机的分发到 wiriterHost 对应的 readhost 执行,writerHost 不负担读压力
             */
            int balance = Integer.valueOf(element.getAttribute("balance"));
            /**
             * 读取切换类型
             * -1 表示不自动切换
             * 1 默认值,自动切换
             * 2 基于MySQL主从同步的状态决定是否切换
             * 心跳询句为 show slave status
             * 3 基于 MySQL galary cluster 的切换机制
             */
            String switchTypeStr = element.getAttribute("switchType");
            int switchType = switchTypeStr.equals("") ? -1 : Integer.valueOf(switchTypeStr);
            //读取从延迟界限
            String slaveThresholdStr = element.getAttribute("slaveThreshold");
            int slaveThreshold = slaveThresholdStr.equals("") ? -1 : Integer.valueOf(slaveThresholdStr);
    
            //如果 tempReadHostAvailable 设置大于 0 则表示写主机如果挂掉, 临时的读服务依然可用
            String tempReadHostAvailableStr = element.getAttribute("tempReadHostAvailable");
            boolean tempReadHostAvailable = tempReadHostAvailableStr.equals("") ? false : Integer.valueOf(tempReadHostAvailableStr) > 0;
            /**
             * 读取 写类型
             * 这里只支持 0 - 所有写操作仅配置的第一个 writeHost
             */
            String writeTypStr = element.getAttribute("writeType");
            int writeType = "".equals(writeTypStr) ? PhysicalDBPool.WRITE_ONLYONE_NODE : Integer.valueOf(writeTypStr);
    
    
            String dbDriver = element.getAttribute("dbDriver");
            String dbType = element.getAttribute("dbType");
            String filters = element.getAttribute("filters");
            String logTimeStr = element.getAttribute("logTime");
            long logTime = "".equals(logTimeStr) ? PhysicalDBPool.LONG_TIME : Long.valueOf(logTimeStr) ;
            //读取心跳语句
            String heartbeatSQL = element.getElementsByTagName("heartbeat").item(0).getTextContent();
            //读取 初始化sql配置,用于oracle
            NodeList connectionInitSqlList = element.getElementsByTagName("connectionInitSql");
            String initConSQL = null;
            if (connectionInitSqlList.getLength() > 0) {
                initConSQL = connectionInitSqlList.item(0).getTextContent();
            }
            //读取writeHost
            NodeList writeNodes = element.getElementsByTagName("writeHost");
            DBHostConfig[] writeDbConfs = new DBHostConfig[writeNodes.getLength()];
            Map<Integer, DBHostConfig[]> readHostsMap = new HashMap<Integer, DBHostConfig[]>(2);
            for (int w = 0; w < writeDbConfs.length; w++) {
                Element writeNode = (Element) writeNodes.item(w);
                writeDbConfs[w] = createDBHostConf(name, writeNode, dbType, dbDriver, maxCon, minCon,filters,logTime);
                NodeList readNodes = writeNode.getElementsByTagName("readHost");
                //读取对应的每一个readHost
                if (readNodes.getLength() != 0) {
                    DBHostConfig[] readDbConfs = new DBHostConfig[readNodes.getLength()];
                    for (int r = 0; r < readDbConfs.length; r++) {
                        Element readNode = (Element) readNodes.item(r);
                        readDbConfs[r] = createDBHostConf(name, readNode, dbType, dbDriver, maxCon, minCon,filters, logTime);
                    }
                    readHostsMap.put(w, readDbConfs);
                }
            }
    
            DataHostConfig hostConf = new DataHostConfig(name, dbType, dbDriver, 
                    writeDbConfs, readHostsMap, switchType, slaveThreshold, tempReadHostAvailable);     
    
            hostConf.setMaxCon(maxCon);
            hostConf.setMinCon(minCon);
            hostConf.setBalance(balance);
            hostConf.setWriteType(writeType);
            hostConf.setHearbeatSQL(heartbeatSQL);
            hostConf.setConnectionInitSql(initConSQL);
            hostConf.setFilters(filters);
            hostConf.setLogTime(logTime);
            dataHosts.put(hostConf.getName(), hostConf);
        }
    }

先读取每个DataHost的通用配置,之后读取每个DataHost对应的writeHost以及每个writeHost对应的readHost。配置好后,保存在:

    private final Map<String, DataHostConfig> dataHosts;

之后读取载入DataHost:
XMLSchemaLoader.java:

    private void loadDataNodes(Element root) {
        //读取DataNode分支
        NodeList list = root.getElementsByTagName("dataNode");
        for (int i = 0, n = list.getLength(); i < n; i++) {
            Element element = (Element) list.item(i);
            String dnNamePre = element.getAttribute("name");
    
            String databaseStr = element.getAttribute("database");
            String host = element.getAttribute("dataHost");
            //字符串不为空
            if (empty(dnNamePre) || empty(databaseStr) || empty(host)) {
                throw new ConfigException("dataNode " + dnNamePre + " define error ,attribute can't be empty");
            }
            //dnNames(name),databases(database),hostStrings(dataHost)都可以配置多个,以',', '$', '-'区分,但是需要保证database的个数*dataHost的个数=name的个数
            //多个dataHost与多个database如果写在一个标签,则每个dataHost拥有所有database
            //例如:<dataNode name="dn1$0-75" dataHost="localhost$1-10" database="db$0-759" />
            //则为:localhost1拥有dn1$0-75,localhost2也拥有dn1$0-75(对应db$76-151)
            String[] dnNames = io.mycat.util.SplitUtil.split(dnNamePre, ',', '$', '-');
            String[] databases = io.mycat.util.SplitUtil.split(databaseStr, ',', '$', '-');
            String[] hostStrings = io.mycat.util.SplitUtil.split(host, ',', '$', '-');
    
            if (dnNames.length > 1 && dnNames.length != databases.length * hostStrings.length) {
                throw new ConfigException("dataNode " + dnNamePre
                                + " define error ,dnNames.length must be=databases.length*hostStrings.length");
            }
            if (dnNames.length > 1) {
    
                List<String[]> mhdList = mergerHostDatabase(hostStrings, databases);
                for (int k = 0; k < dnNames.length; k++) {
                    String[] hd = mhdList.get(k);
                    String dnName = dnNames[k];
                    String databaseName = hd[1];
                    String hostName = hd[0];
                    createDataNode(dnName, databaseName, hostName);
                }
    
            } else {
                createDataNode(dnNamePre, databaseStr, host);
            }
    
        }
    }
    private void createDataNode(String dnName, String database, String host) {
    
        DataNodeConfig conf = new DataNodeConfig(dnName, database, host);       
        if (dataNodes.containsKey(conf.getName())) {
            throw new ConfigException("dataNode " + conf.getName() + " duplicated!");
        }
    
        if (!dataHosts.containsKey(host)) {
            throw new ConfigException("dataNode " + dnName + " reference dataHost:" + host + " not exists!");
        }
        dataNodes.put(conf.getName(), conf);
    }

生成的是DataNode类,放入:

    private final Map<String, DataNodeConfig> dataNodes;
阅读全文
  • 点赞