swoole4.0之打造自己的web开发框架（4）

上一篇：

swoole4.0之打造自己的web开发框架(3)

我们介绍了在协程下使用全局变量的问题以及实现了一个Context，用于协程内的上下文传递，这基本算是和FPM最需要注意的地方了，本篇将实现一个完整的CRUD功能

1、分层

一般我的习惯是代码分为以下几个层

controller， 此层只负责功能的调度，参数的控制，结果的输出

service，此层负责业务逻辑相关的开发

dao，此层负责与数据的交互

entity, 此层定义为数据表的一个映射

2、连接池

在之前的系列文章：swoole4.0 mysql连接池之读写分离有介绍，这里我们直接整合进来就行了。

数据库操作代码：

//file framework/Family/Db/Mysql.php

namespaceFamily\Db;

useFamily\Core\Log;

useSwoole\Coroutine\MySQLasSwMySql;

classMysql

{

/**

*@varMySQL

*/

private$master;//主数据库连接

private$slave;//从数据库连接list

private$config;//数据库配置

/**

*@param$config

*@returnmixed

*@throws\Exception

*@desc连接mysql

*/

public functionconnect($config)

{

//创建主数据连接

$master =newSwMySql();

$res = $master->connect($config['master']);

if($res ===false) {

//连接失败，抛弃常

throw new\Exception($master->connect_error, $master->errno);

}else{

//存入master资源

$this->master= $master;

}

if(!empty($config['slave'])) {

//创建从数据库连接

foreach($config['slave']as$conf) {

$slave =newMySQL();

$res = $slave->connect($conf);

if($res ===false) {

//连接失败，抛弃常

throw new\Exception($slave->connect_error, $slave->errno);

}else{

//存入slave资源

$this->slave[] = $slave;

}

}

}

$this->config= $config;

return$res;

}

/**

*@param$type

*@param$index

*@returnMySQL

*@desc单个数据库重连

*@throws\Exception

*/

public functionreconnect($type, $index)

{

//通过type判断是主还是从

if('master'== $type) {

//创建主数据连接

$master =newSwMySql();

$res = $master->connect($this->config['master']);

if($res ===false) {

//连接失败，抛弃常

throw new\Exception($master->connect_error, $master->errno);

}else{

//更新主库连接

$this->master= $master;

}

return$this->master;

}

if(!empty($this->config['slave'])) {

//创建从数据连接

$slave =newSwMySql();

$res = $slave->connect($this->config['slave'][$index]);

if($res ===false) {

//连接失败，抛弃常

throw new\Exception($slave->connect_error, $slave->errno);

}else{

//更新对应的重库连接

$this->slave[$index] = $slave;

}

return$slave;

}

}

/**

*@param$name

*@param$arguments

*@returnmixed

*@desc利用__call,实现操作mysql,并能做断线重连等相关检测

*@throws\Exception

*/

public function__call($name, $arguments)

{

$sql = $arguments[];

$res = $this->chooseDb($sql);

$db = $res['db'];

// $result = call_user_func_array([$db, $name], $arguments);

$result = $db->$name($sql);

Log::info($sql);

if(false=== $result) {

Log::warning('mysql query false', [$sql]);

if(!$db->connected) {//断线重连

$db = $this->reconnect($res['type'], $res['index']);

Log::info('mysql reconnect', $res);

$result = $db->$name($sql);

return$this->parseResult($result, $db);

}

if(!empty($db->errno)) {//有错误码，则抛出弃常

throw new\Exception($db->error, $db->errno);

}

}

return$this->parseResult($result, $db);

}

/**

*@param$result

*@param$db MySQL

*@returnarray

*@desc格式化返回结果：查询：返回结果集，插入：返回新增id, 更新删除等操作：返回影响行数

*/

public functionparseResult($result, $db)

{

if($result ===true) {

return[

'affected_rows'=> $db->affected_rows,

'insert_id'=> $db->insert_id,

];

}

return$result;

}

/**

*@param$sql

*@desc根据sql语句，选择主还是从

* @ 判断有select 则选择从库， insert, update, delete等选择主库

*@returnarray

*/

protected functionchooseDb($sql)

{

if(!empty($this->slave)) {

//查询语句，随机选择一个从库

if('select'==strtolower(substr($sql,,6))) {

if(1==count($this->slave)) {

$index =;

}else{

$index =array_rand($this->slave);

}

return[

'type'=>'slave',

'index'=> $index,

'db'=> $this->slave[$index],

];

}

}

return[

'type'=>'master',

'index'=>,

'db'=> $this->master

];

}

}

连接池代码：

//file framework/Family/Pool/Mysql.php

namespaceFamily\Pool;

useFamily\Db\MysqlasDB;

usechan;

classMysql

{

private static$instance;

private$pool;//连接池容器，一个channel

private$config;

/**

*@paramnull $config

*@returnMysql

*@desc获取连接池实例

*@throws\Exception

*/

public static functiongetInstance($config =null)

{

if(empty(self::$instance)) {

if(empty($config)) {

throw new\Exception("mysql config empty");

}

self::$instance=new static($config);

}

return self::$instance;

}

/**

* Mysql constructor.

*@param$config

*@throws\Exception

*@desc初始化，自动创建实例,需要放在workerstart中执行

*/

public function__construct($config)

{

if(empty($this->pool)) {

$this->config= $config;

$this->pool=newchan($config['pool_size']);

for($i =; $i

$mysql =newDB();

$res = $mysql->connect($config);

if($res ==false) {

//连接失败，抛弃常

throw new\Exception("failed to connect mysql server.");

}else{

//mysql连接存入channel

$this->put($mysql);

}

}

}

}

/**

*@param$mysql

*@desc放入一个mysql连接入池

*/

public functionput($mysql)

{

$this->pool->push($mysql);

}

/**

*@returnmixed

*@desc获取一个连接，当超时，返回一个异常

*@throws\Exception

*/

public functionget()

{

$mysql = $this->pool->pop($this->config['pool_get_timeout']);

var_dump($mysql);

if(false=== $mysql) {

throw new\Exception("get mysql timeout, all mysql connection is used");

}

return$mysql;

}

/**

*@returnmixed

*@desc获取当时连接池可用对象

*/

public functiongetLength()

{

return$this->pool->length();

}

}

application/config/default.php 加上数据库配置

'mysql'=> [

'pool_size'=>3,//连接池大小

'pool_get_timeout'=>0.5,//当在此时间内未获得到一个连接，会立即返回。（表示所以的连接都已在使用中）

'master'=> [

'host'=>'127.0.0.1',//数据库ip

'port'=>3306,//数据库端口

'user'=>'root',//数据库用户名

'password'=>'123456',//数据库密码

'database'=>'test',//默认数据库名

'timeout'=>0.5,//数据库连接超时时间

'charset'=>'utf8mb4',//默认字符集

'strict_type'=>true,//ture，会自动表数字转为int类型

],

],

在Family/run方法里，初始化连接池

$http->on('workerStart',function(\swoole_http_server $serv, int $worker_id) {

if(function_exists('opcache_reset')) {

//清除opcache 缓存，swoole模式下其实可以关闭opcache

\opcache_reset();

}

try{

$mysqlConfig =Config::get('mysql');

if(!empty($mysqlConfig)) {

//配置了mysql, 初始化mysql连接池

Pool\Mysql::getInstance($mysqlConfig);

}

}catch(\Exception $e) {

//初始化异常，关闭服务

print_r($e);

$serv->shutdown();

}catch(\Throwable $throwable) {

//初始化异常，关闭服务

print_r($throwable);

$serv->shutdown();

}

});

框架层的代码就完成，下面接着看业务逻辑

我们创建一个user表，用于做CRUD的操作表

3、entity定义

entity是表的映射，所以这个代码简单：

//file: application/entity/user.phU

namespaceentity;

useFamily\MVC\Entity;

classUserextendsEntity

{

/**

* 对应的数据库表名

*/

constTABLE_NAME='user';

/**

* 主键字段名

*/

constPK_ID='id';

//以下对应的数据库字段名

public$id;

public$name;

public$password;

}

这里有个Entity基类，主要目的是: mysql查询回来的数据是一个数组，我们能够把数据自动填充到entity里来，代码如下：

//file framework/Family/MVC/Entity.php

namespaceFamily\MVC;

classEntity

{

/**

* Entity constructor.

*@paramarray $array

*@desc把数组填充到entity

*/

public function__construct(array$array)

{

if(empty($array)) {

return$this;

}

foreach($arrayas$key => $value) {

if(property_exists($this, $key)) {

$this->$key= $value;

}

}

}

}

4、Dao封装

dao层是与数据库的交互，这个我们可以抽离出一个基类，把常见的crud操作封封起来, 代码如下：

//file framework/Family/MVC/Dao.php

namespaceFamily\MVC;

useFamily\Db\Mysql;

useFamily\Pool\MysqlasMysqlPool;

useFamily\Coroutine\Coroutine;

classDao

{

/**

*@varentity名

*/

private$entity;

/**

*@varmysql连接数组

*@desc不同协程不能复用mysql连接，所以通过协程id进行资源隔离

*/

private$dbs;

/**

*@varMysql

*/

private$db;

//表名

private$table;

//主键字段名

private$pkId;

public function__construct($entity)

{

$this->entity= $entity;

$coId =Coroutine::getId();

if(empty($this->dbs[$coId])) {

//不同协程不能复用mysql连接，所以通过协程id进行资源隔离

//达到同一协程只用一个mysql连接，不同协程用不同的mysql连接

$this->dbs[$coId] =MysqlPool::getInstance()->get();

$entityRef =new\ReflectionClass($this->entity);

$this->table= $entityRef->getConstant('TABLE_NAME');

$this->pkId= $entityRef->getConstant('PK_ID');

defer(function() {

//利用协程的defer特性，自动回收资源

$this->recycle();

});

}

$this->db= $this->dbs[$coId];

}

/**

*@throws\Exception

*@descmysql资源回收到连接池

*/

public functionrecycle()

{

$coId =Coroutine::getId();

if(!empty($this->dbs[$coId])) {

$mysql = $this->dbs[$coId];

MysqlPool::getInstance()->put($mysql);

unset($this->dbs[$coId]);

}

}

/**

*@returnmixed

*@desc获取表名

*/

public functiongetLibName()

{

return$this->table;

}

/**

*@param$id

*@paramstring $fields

*@returnmixed

*@desc通过主键查询记录

*/

public functionfetchById($id, $fields ='*')

{

return$this->fetchEntity("{$this->pkId}={$id}", $fields);

}

/**

*@paramstring $where

*@paramstring $fields

*@paramnull $orderBy

*@returnmixed

*@desc通过条件查询一条记录，并返回一个entity

*/

public functionfetchEntity($where ='1', $fields ='*', $orderBy =null)

{

$result = $this->fetchArray($where, $fields, $orderBy,1);

if(!empty($result[])) {

return new$this->entity($result[]);

}

return null;

}

/**

*@paramstring $where

*@paramstring $fields

*@paramnull $orderBy

*@paramint $limit

*@returnmixed

*@desc通过条件查询记录列表，并返回entity列表

*/

public functionfetchAll($where ='1', $fields ='*', $orderBy =null, $limit =)

{

$result = $this->fetchArray($where, $fields, $orderBy, $limit);

if(empty($result)) {

return$result;

}

foreach($resultas$index => $value) {

$result[$index] =new$this->entity($value);

}

return$result;

}

/**

*@paramstring $where

*@paramstring $fields

*@paramnull $orderBy

*@paramint $limit

*@returnmixed

*@desc通过条件查询

*/

public functionfetchArray($where ='1', $fields ='*', $orderBy =null, $limit =)

{

$query ="SELECT{$fields}FROM{$this->getLibName()}WHERE{$where}";

if($orderBy) {

$query .=" order by{$orderBy}";

}

if($limit) {

$query .=" limit{$limit}";

}

return$this->db->query($query);

}

/**

*@paramarray $array

*@returnbool

*@desc插入一条记录

*/

public functionadd(array$array)

{

$strFields ='`'.implode('`,`',array_keys($array)) .'`';

$strValues ="'".implode("','",array_values($array)) ."'";

$query ="INSERT INTO{$this->getLibName()}({$strFields}) VALUES ({$strValues})";

if(!empty($onDuplicate)) {

$query .='ON DUPLICATE KEY UPDATE '. $onDuplicate;

}

echo$query .PHP_EOL;

$result = $this->db->query($query);

if(!empty($result['insert_id'])) {

return$result['insert_id'];

}

return false;

}

/**

*@paramarray $array

*@param$where

*@returnbool

*@throws\Exception

*@desc按条件更新记录

*/

public functionupdate(array$array, $where)

{

if(empty($where)) {

throw new\Exception('update 必需有where条件限定');

}

$strUpdateFields ='';

foreach($arrayas$key => $value) {

$strUpdateFields .="`{$key}` = '{$value}',";

}

$strUpdateFields =rtrim($strUpdateFields,',');

$query ="UPDATE{$this->getLibName()}SET{$strUpdateFields}WHERE{$where}";

echo$query;

$result = $this->db->query($query);

return$result['affected_rows'];

}

/**

*@param$where

*@returnmixed

*@throws\Exception

*@desc按条件删除记录

*/

public functiondelete($where)

{

if(empty($where)) {

throw new\Exception('delete 必需有where条件限定');

}

$query ="DELETE FROM{$this->getLibName()}WHERE{$where}";

$result = $this->db->query($query);

return$result['affected_rows'];

}

}

这里有个需要注意的地方就是，不同的协程不能复用一个mysql连接，但同一协程内可以复用，所以在Dao类做了一些优化处理

5、业务DAO实现

有了dao的基类，我们的业务dao实现就非常简单的了，代码如下：

//file application/dao/User.php

namespacedao;

useFamily\MVC\Dao;

useFamily\Core\Singleton;

classUserextendsDao

{

useSingleton;

public function__construct()

{

parent::__construct('\entity\User');

}

}

这里有个

traitSingleton

作用是代码复用，因为实现单例的方法都一样，代码如下：

//file frame/Family/Core/Singleton.php

namespaceFamily\Core;

traitSingleton

{

private static$instance;

static functiongetInstance(...$args)

{

if(!isset(self::$instance)) {

self::$instance=new static(...$args);

}

return self::$instance;

}

}

6、业务Service实现

业务逻辑也简单，实现了几个功能，代码如下：

//file application/service/User.php

namespaceservice;

usedao\UserasUserDao;

useFamily\Core\Singleton;

classUser

{

useSingleton;

/**

*@param$id

*@returnmixed

*@desc通过uid查询用户信息

*/

public functiongetUserInfoByUId($id)

{

returnUserDao::getInstance()->fetchById($id);

}

/**

*@returnmixed

*@desc获取所有用户列表

*/

public functiongetUserInfoList()

{

returnUserDao::getInstance()->fetchAll();

}

/**

*@paramarray $array

*@returnbool

*@desc添加一个用户

*/

public functionadd(array$array)

{

returnUserDao::getInstance()->add($array);

}

/**

*@paramarray $array

*@param$id

*@returnbool

*@throws\Exception

*@desc按id更新一个用户

*/

public functionupdateById(array$array, $id)

{

returnUserDao::getInstance()->update($array,"id={$id}");

}

/**

*@param$id

*@returnmixed

*@throws\Exception

*@desc按id删除用户

*/

public functiondeleteById($id)

{

returnUserDao::getInstance()->delete("id={$id}");

}

}

7、Controller实现

最后就是实现controller了，代码也很简单，如下：

//file: application/controller/Index.php

namespacecontroller;

useFamily\MVC\Controller;

useservice\UserasUserService;

classIndexextendsController

{

public functionindex()

{

return'i am family by route!'.json_encode($this->request->get);

}

public functiontong()

{

return'i am tong ge';

}

/**

*@returnfalse|string

*@throws\Exception

*@desc返回一个用户信息

*/

public functionuser()

{

if(empty($this->request->get['uid'])) {

throw new\Exception("uid 不能为空 ");

}

$result =UserService::getInstance()->getUserInfoByUId($this->request->get['uid']);

returnjson_encode($result);

}

/**

*@returnfalse|string

*@desc返回用户列表

*/

public function list()

{

$result =UserService::getInstance()->getUserInfoList();

returnjson_encode($result);

}

/**

*@returnbool

*@desc添加用户

*/

public functionadd()

{

$array = [

'name'=> $this->request->get['name'],

'password'=> $this->request->get['password'],

];

returnUserService::getInstance()->add($array);

}

/**

*@returnbool

*@throws\Exception

*@desc更新用户信息

*/

public functionupdate()

{

$array = [

'name'=> $this->request->get['name'],

'password'=> $this->request->get['password'],

];

$id = $this->request->get['id'];

returnUserService::getInstance()->updateById($array, $id);

}

/**

*@returnmixed

*@throws\Exception

*@desc删除用户信息

*/

public functiondelete()

{

$id = $this->request->get['id'];

returnUserService::getInstance()->deleteById($id);

}

}

这里也把controller一些操作抽出成一个基类，目前只是把request对象抽出来了，后续如：参数的获取和判断，安全过滤等等，都可以在基类里补充，代码如下：

//file framework/Family/MVC/Controller.php

namespaceFamily\MVC;

useFamily\Pool\Context;

classController

{

protected$request;

public function__construct()

{

//通过context拿到$request, 再也不用担收数据错乱了

$context =Context::getContext();

$this->request= $context->getRequest();

}

}

8、运行

http://127.0.0.1:9501/Index/add?name=shenzhe5&password=555

http://127.0.0.1:9501/Index/list

http://127.0.0.1:9501/Index/update?name=shenzhe0&password=555&id=5

http://127.0.0.1:9501/Index/delete?id=5

至此，我们就拥有一个完整的CRUD的操作，并且把MVC的基础功能封装在框架里了

可以看到，代码的思路和习惯和常用的fpm web开发框架几无差别

下一篇：我们将拥抱composer, 并通过整合一个composer router包，把router功能升级，有了composer，我们将有了取之不尽的各类库，框架的能力也将大大的得到拓展