MySQL集群读写分离的自定义实现

基于MySQL Router能够实现高可用，读写分离，负载均衡之类的，MySQL Router能够说是很是轻量级的一个中间件了。
看了一下MySQL Router的原理，其实并不复杂，原理也并不难理解，其实就是一个相似于VIP的代理功能，其中一个MySQL Router有两个端口号，分别是对读和写的转发。
至于选择哪一个端口号，须要在申请链接的时候自定义选择，换句话说就是在生成链接字符串的时候，要指明是读操做仍是写操做，而后由MySQL Router转发到具体的服务器上。

引用这里的话说就是：
通常来讲，经过不一样端口实现读/写分离，并不是好方法，最大的缘由是须要在应用程序代码中指定这些链接端口。
可是，MySQL Router只能经过这种方式实现读写分离，因此MySQL Router拿来当玩具玩玩就好。其原理参考下图，相关安装配置等很是简单。

　

其实暂不论“MySQL Router拿来当玩具玩玩就好”，相似须要本身指定端口（或者说指定读写）来实现读写分离这种方式，本身彻底能够实现，又何须用一个中间件呢？
对于MySQL Router来讲，它本身自己又是单点的，还要考虑Router自身的高可用（解决了一个问题的同时又引入一个问题）。
很早以前就在想，可不能够尝试不借助中间件，也就无需关注中间件自身的高可用，本身实现读写分离呢？

对于最简单的master-salve复制的集群方式的读写分离，
能够集群中的不一样节点指定不一样的优先级，把master服务器的优先级指定到最高，其他两个指定成一个较低的优先级
对于应用程序发起的请求，须要指明是读仍是写，若是是写操做，就指定到master上执行，若是是读操做，就随机地指向slave操做，彻底能够在链接层就实现相似于MySQL Router的功能。
其实很是简单，花不了多久就能够实现相似这么一个功能，在链接层实现读写分离，高可用，负载均衡，demo一个代码实现。

 

 

以下简单从数据库链接层实现了读写分离以及负载均衡。
1，写请求指向链接字符串中最高优先级的master，若是指定的最高优先级实例不可用，这里假如是实现了故障转移，依次寻找次优先级的实例
2，slave复制master的数据，读请求随机指向不一样的slave，一旦某个slave不可用，继续寻找其余的slave
3，维护一个链接池，链接一概从链接池中获取。

故障转移能够独立实现，不须要在链接层作，链接层也不是作故障转移的。这样一旦发生故障，只要实现了故障转移，应用程序端能够不用作任何修改。

# -*- coding: utf-8 -*-
import pymysql
import random
from DBUtils.PooledDB import PooledDB
import socket


class MySQLRouter:

    operation = None
    conn_list = []

    def __init__(self, *args, **kwargs):
        for k, v in kwargs.items():
            setattr(self, k, v)

    # 探测实例端口号
    @staticmethod
    def get_mysqlservice_status(host,port):
        mysql_stat = None
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        result = s.connect_ex((host, int(port)))
        # port os open
        if (result == 0):
            mysql_stat = 1
        return mysql_stat

    def get_connection(self):
        if not conn_list:
            raise("no config error")

        conn = None
        current_conn = None
        # 依据节点优先级排序
        self.conn_list.sort(key=lambda k: (k.get('priority', 0)))
        #写或者未定义请求，一概指向高优先级的服务器，可读写
        if(self.operation.lower() == "write") or not self.operation:
            for conn in conn_list:
                # 若是最高优先级的主节点不可达，这里假设成功实现了故障转移，继续找次优先级的实例。
                if self.get_mysqlservice_status(conn["host"], conn["port"]):
                    current_conn = conn
                    break
                else:
                    continue
        #读请求随机指向不一样的slave
        elif(self.operation.lower() == "read"):
            #随机获取除了最该优先级节点以外的节点
            conn_read_list = conn_list[1:len(conn_list)]
            random.shuffle(conn_read_list)
            for conn in conn_read_list:
                #若是不可达，继续寻找其余除了主节点以外的节点
                if self.get_mysqlservice_status(conn["host"], conn["port"]):
                    current_conn = conn
                    break
                else:
                    continue
        try:
            #从链接池中获取当前链接
            if (current_conn):
                pool = PooledDB(pymysql,20, host=current_conn["host"], port=current_conn["port"], user=current_conn["user"], password=current_conn["password"],db=current_conn["database"])
                conn = pool.connection()
        except:
            raise

        if not conn:
            raise("create connection error")

        return conn;


if __name__ == '__main__':

    #定义三个实例
    conn_1 = {'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'root',"database":"db01","priority":100}
    conn_2 = {'host': '127.0.0.1', 'port': 3307, 'user': 'root', 'password': 'root',"database":"db01","priority":200}
    conn_3 = {'host': '127.0.0.1', 'port': 3308, 'user': 'root', 'password': 'root',"database":"db01","priority":300}
    
    conn_list = []
    conn_list.append(conn_1)
    conn_list.append(conn_2)
    conn_list.append(conn_3)

    print("####execute update on master####")
    myrouter = MySQLRouter(conn_list=conn_list, operation="write")
    conn = myrouter.get_connection()
    cursor = conn.cursor()
    cursor.execute("update t01 set update_date = now() where id = 1")
    conn.commit()
    cursor.close()
    conn.close()

    print("####loop execute read on slave,query result####")
    #循环读，判断读指向哪一个节点。
    for loop in range(10):
        myrouter = MySQLRouter(conn_list = conn_list,operation = "read")
        conn = myrouter.get_connection()
        cursor = conn.cursor()
        cursor.execute("SELECT id,cast(update_date as char), CONCAT('instance port is: ', CAST( @@PORT AS CHAR)) AS port FROM t01;")
        result = cursor.fetchone()
        print(result)
        cursor.close()
        conn.close()

这里用过服务器的一个优先级，将写请求指向最高优先级的master服务器，读请求随机指向非最高优先级的slave，
对于更新请求，都在master上执行，slave复制了master的数据，每次读到的数据都不同，而且每次都请求的执行，基本上都随机地指向了两台slave服务器
经过查询返回一个端口号，来判断读请求是否平均分散到了不通的slave端。

与“MySQL Router拿来当玩具玩玩就好”相比，这里的实现同样low，由于对数据的请求须要请求明确指定是读仍是写。

不过，对于自动读写分离，无非是一个SQL语句执行的是的读或写判断问题，并不是难事，这个须要解析请求的SQL是读的仍是写的问题。
某些数据库中间件能够实现自动的读写分离，可是要明白，对于那些支持自动读写分离的中间件，每每是要受到必定的约束的，好比不能用存储过程什么的，为何呢？
仍是上面提到的SQL解析的问题，由于一旦使用了存储过程，没法解析出来这个SQL究竟是执行的是读仍是写，最起码不是太直接。
对于SQL读写的判断，也就是维护一个读或者写的枚举正则表达式，非读即写，只是要格外关注这个读写的判断的效率问题。