docker 部署读写分离及laravel读写分离原理剖析

Posted 2021-04-21 PHP纯干货

mysql主从环境配置

拉取mysql docker镜像

docker pull mysql:5.7

MySQL配置主从

# master.cnf[mysqld]
server-id = 1log-bin = mysql-bin# slave.cnf[mysqld]
server-id = 2

启动主服务器

docker run -d -p 50000:3306  -e MYSQL_ROOT_PASSWORD=123456 --name=master mysql:5.7docker cp master.cnf master:/etc/mysql/conf.d

启动从服务器

docker run -d -p50001:3306 -v slave.cnf:/etc/mysql/conf.d  -e MYSQL_ROOT_PASSWORD=123456 --name=slave mysql:5.7docker cp slave.cnf slave:/etc/mysql/conf.d

重启主重容器

docker restart master slave

主容器创建同步账号

docker exec -it master bash
GRANT REPLICATION SLAVE ON *.* to 'mmf'@'%' identified by '123456';

从容器配置

docker exec -it slave bash

change master to
master_host='172.18.0.1',
master_user='mmf',
master_log_file='mysql-bin.000001', //master对应的file
master_log_pos=12610,   //master对应的Position
master_port=50000,      //master端口
master_password='123456';

start slave;

建库,表

create database mmf;

use mmf;

CREATE TABLE `account` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '账号ID',
  `name` varchar(255) NOT NULL COMMENT '昵称',
  `password` varchar(244) NOT NULL DEFAULT ''  COMMENT '密码',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COMMENT='账号表';

laravel如何实现读写分离

读写库的连接

数据库配置

'connections' => [        
    'mysql' => [            
        'driver'    => 'mysql',            
        'host'      => env('DB_LITE_HOST', '127.0.0.1'),            'port'      => env('DB_LITE_PORT', '3306'),            'database'  => env('DB_LITE_DATABASE', 'forge'),            'username'  => env('DB_LITE_USERNAME', 'forge'),            'password'  => env('DB_LITE_PASSWORD', ''),            'charset'   => 'utf8',            'collation' => 'utf8_general_ci',            'prefix'    => '',            'strict'    => false,            'engine'    => 'innodb',            'options'=>[
                PDO::ATTR_STRINGIFY_FETCHES => false,
            ],            
        //读写配置
            'read'  =>  [                
                'port'      => '50001',
            ],            
            'write'  =>  [                
                'port'      => '50000',
            ],
        ],
    ],

laravel读写连接实现

//vendor/laravel/framework/src/Illuminate/Database/DatabaseServiceProvider.php//注册连接服务 
protected function registerConnectionServices(){    
    // The connection factory is used to create the actual connection instances on
    // the database. We will inject the factory into the manager so that it may
    // make the connections while they are actually needed and not of before.

    //解析数据库配置,创建数据库读写连接,
    $this->app->singleton('db.factory', function ($app) {
            return new ConnectionFactory($app);
    });    // The database manager is used to resolve various connections, since multiple
    // connections might be managed. It also implements the connection resolver
    // interface which may be used by other components requiring connections.

    //保存从db.factory创建的读写连接,同一保存与管理
    $this->app->singleton('db', function ($app) {        
            return new DatabaseManager($app, $app['db.factory']);
    });    
            $this->app->bind('db.connection', function ($app) {        
            return $app['db']->connection();
    });
}
            
//vendor/laravel/framework/src/Illuminate/Database/Connectors/ConnectionFactory.php//建立读写连接
public function make(array $config, $name = null){    //解析配置
    $config = $this->parseConfig($config, $name);    //如果设置了读服务器,创建读写服务器连接
    if (isset($config['read'])) {        
        return $this->createReadWriteConnection($config);
    }    //否则创建单连接
    return $this->createSingleConnection($config);
}
        
//建立读写连接
protected function createReadWriteConnection(array $config){    //建立写连接
    $connection = $this->createSingleConnection($this->getWriteConfig($config));    //建立读连接
    return $connection->setReadPdo($this->createReadPdo($config));
}

//读写库的配置
protected function getReadWriteConfig(array $config, $type){    
//如果读写配置了多个,随机取一个
    return isset($config[$type][0])
                    ? Arr::random($config[$type])
                    : $config[$type];
}
//保存读链接在readPdo属性
public function setReadPdo($pdo){    
    $this->readPdo = $pdo;    
    return $this;
}

验证

Route::get('/read-write', function () {    //查看连接
   dd(app('db.connection'));
})->name('wx_auth_success');

连接正常

读写库分配规则

laravel代码实现

insert update delete 使用写库, select 使用读库

1. 原生sql

文件:vendor/laravel/framework/src/Illuminate/Database/Connection.php

//insert
public function insert($query, $bindings = []){    
//insert使用读库
    return $this->statement($query, $bindings);
}
//update
public function update($query, $bindings = []){     
    //update使用读库
    return $this->affectingStatement($query, $bindings);
}
//deletepublic function delete($query, $bindings = []){     
    //delete使用读库
    return $this->affectingStatement($query, $bindings);
}
//select
public function select($query, $bindings = [], $useReadPdo = true){    
    return $this->run($query, $bindings, function ($query, $bindings) use ($useReadPdo) {        
    if ($this->pretending()) {            
            return [];
    }        
        // For select statements, we'll simply execute the query and return an array
        // of the database result set. Each element in the array will be a single
        // row from the database table, and will either be an array or objects.
        //这里拿读库
        $statement = $this->prepared($this->getPdoForSelect($useReadPdo)                          
                    ->prepare($query));        
       $this->bindValues($statement, $this->prepareBindings($bindings));        
        $statement->execute();        
        return $statement->fetchAll();
    });
}
        
protected function getPdoForSelect($useReadPdo = true){    
    //默认 $useReadPdo=true  没有读写分离拿写库,
    //读写分离从getReadPdo()方法那拿取
    return $useReadPdo ? $this->getReadPdo() : $this->getPdo();
} 

//拿读库
public function getReadPdo(){    
    //transactions 后面再讲
    if ($this->transactions >= 1) {        
        return $this->getPdo();
    }    
    if ($this->readPdo instanceof Closure) {        
        return $this->readPdo = call_user_func($this->readPdo);
    }    
    //如果读库连接没有默认还是拿写库
    return $this->readPdo ?: $this->getPdo();
}

1. 查询构建器实现

文件:vendor/laravel/framework/src/Illuminate/Database/Query/Builder.php

Query/Builder构造器封装好参数调用原生Collectio进行crud

//构照方法
public function __construct(ConnectionInterface $connection,                                Grammar $grammar = null,                                Processor $processor = null){    
    $this->connection = $connection; //注入原生Connection类
    $this->grammar = $grammar ?: $connection->getQueryGrammar();    
    $this->processor = $processor ?: $connection->getPostProcessor();
}
//查询构建器的insert,update,delete都是调用Connection的方法实现//insert
 public function insert(array $values){    
    // Since every insert gets treated like a batch insert, we will make sure the
    // bindings are structured in a way that is convenient when building these
    // inserts statements by verifying these elements are actually an array.
    if (empty($values)) {        
        return true;
    }    
    if (! is_array(reset($values))) {        
        $values = [$values];
    }    
    // Here, we will sort the insert keys for every record so that each insert is
    // in the same order for the record. We need to make sure this is the case
    // so there are not any errors or problems when inserting these records.
    else {        
        foreach ($values as $key => $value) {
            ksort($value);            
            $values[$key] = $value;
        }
    }    
    // Finally, we will run this query against the database connection and return
    // the results. We will need to also flatten these bindings before running
    // the query so they are all in one huge, flattened array for execution.
    //插入方法调用Connection对象实现,同原生insert一样
    return $this->connection->insert(        
        $this->grammar->compileInsert($this, $values),        
        $this->cleanBindings(Arr::flatten($values, 1))
    );
}
//update
public function update(array $values){    
    $sql = $this->grammar->compileUpdate($this, $values);    
    return $this->connection->update($sql, $this->cleanBindings(        
            $this->grammar->prepareBindingsForUpdate($this->bindings, $values)
    ));
}
//delete
 public function delete($id = null){    
    // If an ID is passed to the method, we will set the where clause to check the
    // ID to let developers to simply and quickly remove a single row from this
    // database without manually specifying the "where" clauses on the query.
    if (! is_null($id)) {        
        $this->where($this->from.'.id', '=', $id);
    }    
    return $this->connection->delete(        
        $this->grammar->compileDelete($this), $this->getBindings()
    );
}
//对于select
    public function get($columns = ['*']){    
        $original = $this->columns;    
        if (is_null($original)) {        
            $this->columns = $columns;
    }    
     $results = $this->processor->processSelect($this, $this->runSelect());
     //select此处处理
    $this->columns = $original;    
    return collect($results);
} 
    
protected function runSelect(){    
    //select同原生Connection一样
    return $this->connection->select(        
        $this->toSql(), $this->getBindings(), ! $this->useWritePdo //默认不使用写pdo
    );
}
//如果希望手动指定写库可以调用下面这个方法
public function useWritePdo(){    
    $this->useWritePdo = true;    
    return $this;
}

1. ORM查询

文件:vendor/laravel/framework/src/Illuminate/Database/Eloquent/Model.php

ORM的增删改查其实是调用的Eloquent/Builder

Eloquent/Builder 调用Query/Builder类的方法, Query/Builder使用Connection.php连接模型

本质上和查询构造器一样

//在类中没有的方法,调用魔术方法

public static function __callStatic($method, $parameters){    
    return (new static)->$method(...$parameters);
}

public function __call($method, $parameters){    
    if (in_array($method, ['increment', 'decrement'])) {        
        return $this->$method(...$parameters);
    }    
    //从newQuery方法获取Builder实例
    return $this->newQuery()->$method(...$parameters);
}
    
//query方法获取

public static function query(){    
    return (new static)->newQuery();
}
    
//返回 Eloquent/Builder实例

public function newQuery(){    
    $builder = $this->newQueryWithoutScopes();    
    foreach ($this->getGlobalScopes() as $identifier => $scope) {        
        $builder->withGlobalScope($identifier, $scope);
    }    
    return $builder;
}
    
//将Query/Builder实例保存在Eloquent/Builder实例中

public function newQueryWithoutScopes(){    
    $builder = $this->newEloquentBuilder($this->newBaseQueryBuilder());    
    // Once we have the query builders, we will set the model instances so the
    // builder can easily access any information it may need from the model
    // while it is constructing and executing various queries against it.
    return $builder->setModel($this)                
        ->with($this->with)                
        ->withCount($this->withCount);
}
        
//返回Query/Builder实例
 protected function newBaseQueryBuilder(){    
    $connection = $this->getConnection();    
    return new QueryBuilder(        
        $connection, $connection->getQueryGrammar(), $connection->getPostProcessor()
    );
}

//Eloquent/Builder
public function __construct(QueryBuilder $query){    
    $this->query = $query;
}
//指定写库读
public static function onWriteConnection(){    
    $instance = new static;    
    return $instance->newQuery()->useWritePdo();
}
//get
public function get($columns = ['*']){    
    //scope搜索
    $builder = $this->applyScopes();    
    // If we actually found models we will also eager load any relationships that
    // have been specified as needing to be eager loaded, which will solve the
    // n+1 query issue for the developers to avoid running a lot of queries.
    if (count($models = $builder->getModels($columns)) > 0 //获取数据) 
    {
        $models = $builder->eagerLoadRelations($models);//渴求加载
    }    
    return $builder->getModel()->newCollection($models);
}
//get数据
public function getModels($columns = ['*']){    
    return $this->model->hydrate(        
        $this->query->get($columns)->all() //query/builder get()数据.
    )->all();
}

//delete
public function delete(){    
    if (isset($this->onDelete)) {        
        return call_user_func($this->onDelete, $this);
    }    
        /*    public function toBase()    {        return $this->applyScopes()->getQuery();    }    */
    return $this->toBase()->delete();
    //query/builder delete()}

增删改查连库测试

事务解决方案

对于在同一事务中,如果select拿读库,insert拿写库,insert在主库中的数据在从库中找不到,肯定是会有问题的,laravel是如何解决这一问题的呢?

//vendor/laravel/framework/src/Illuminate/Database/Concerns/ManagesTransactions.php//开启事务

public function transaction(Closure $callback, $attempts = 1){    for ($currentAttempt = 1; $currentAttempt <= $attempts; $currentAttempt++) {        
        //开启事务
        $this->beginTransaction();        
        // We'll simply execute the given callback within a try / catch block and if we
        // catch any exception we can rollback this transaction so that none of this
        // gets actually persisted to a database or stored in a permanent fashion.
        try {            
            return tap($callback($this), function ($result) {                
                $this->commit();
            });
        }        
        // If we catch an exception we'll rollback this transaction and try again if we
        // are not out of attempts. If we are out of attempts we will just throw the
        // exception back out and let the developer handle an uncaught exceptions.
        catch (Exception $e) {            
            $this->handleTransactionException(                
                $e, $currentAttempt, $attempts
            );
        } catch (Throwable $e) {            
                $this->rollBack();            
                throw $e;
        }
}
//启动事务//当前会话自增
// protected $transactions = 0;默认会话为0
//在上一段中依稀有这样的代码/* if ($this->transactions >= 1) {    return $this->getPdo(); } */
//如果开启的会话大于1,就使用写库,这也就是laravel解决事务读写分离的关键

public function beginTransaction(){    
    $this->createTransaction();

    ++$this->transactions;//当前会话自增

    $this->fireConnectionEvent('beganTransaction');
}
protected function createTransaction(){        
    if ($this->transactions == 0) {            
        try {                
            //在写库中开启事务
                $this->getPdo()->beginTransaction();
            } catch (Exception $e) {                
                $this->handleBeginTransactionException($e);
            }
        } elseif ($this->transactions >= 1 && $this->queryGrammar->supportsSavepoints()) {            
                $this->createSavepoint();
        }
    }

事务验证