投票链接制作哪家服务好_校园网网络规划与设计方案_域名关键词排名查询_拼多多跨境电商平台

引言 thinkphp* 大道至简

负载均衡

分布式服务

一、读写分离

为什么要读写分离
理论上来说读写请求不要超过2000/s，如果加了缓存之后，到数据库请求还是超过2000以上考虑读写分离
使得读请求可以在不同机器并发，用了读写分离之后可以通过动态扩展读服务器增加读效率，这与redis中的主从架构读写分离、copyOnWrite机制的并发容器、以及数据库MVCC机制有点相识，都是通过读请求的数据备份增加读写并发效率。
适用于业务场景中，读请求大于写请求的情况，读写分离使得系统能够更多的容纳读请求并发

1、读写分离的实现方式

一般来说是基于mysql自带的主从复制功能。mysql主从复制的流程图如下：

总结mysql的主从复制过程大体是主库有一个进程专门是将将记录的Binlog日志发送到从库，从库有一个io线程（mysql5.6.x之后IO线程可以多线程写入relay日志）将收到的数据写入relay日志当中，另外还有一个SQL进程专门读取relay日志，根据relay日志重做命令(mysql5.7版本之后，从可以并行读取relay log重放命令（按库并行，每个库一个线程）)。

2、主从同步的三种模式

2-1、异步模式（mysql async-mode）

异步模式如下图所示，这种模式下，主节点不会主动push bin log到从节点，这样有可能导致failover的情况下，也许从节点没有即时地将最新的bin log同步到本地。

2-2、半同步模式（mysql semi-sync）

这种模式下主节点只需要接收到其中一台从节点的返回信息，就会commit；否则需要等待直到超时时间然后切换成异步模式再提交；这样做的目的可以使主从数据库的数据延迟缩小，可以提高数据安全性，确保了事务提交后，binlog至少传输到了一个从节点上，不能保证从节点将此事务更新到db中。性能上会有一定的降低，响应时间会变长

2-3、同步模式（mysql semi-sync）

全同步模式是指主节点和从节点全部执行了commit并确认才会向客户端返回成功。

2-4 总结

在代码中插入之后，又查询这样的操作是不可靠，可能导致插入之后，查出来的时候还没有同步到从库，所以查出来为null。如何应对这种情况了？其实并不能从根本上解决这种情况的方案。只能一定程度通过降低主从延迟来尽量避免。

降低主从延迟的方法有：
拆主库，降低主库并发，降低主库并发，此时主从延迟可以忽略不计，但并不能保证一定不会出现上述情况。
打开并行复制-但这个效果一般不大，因为写入数据可能只针对某个库并发高，而mysql的并行粒度并不小，是以库为粒度的。
但这并不能根本性解决这个问题，其实面对这种情况最好的处理方式是：
重写代码，插入之后不要更新
如果确实是存在先插入，立马就能查询到，然后立马执行一些操作，那么可以对这个查询设置直连主库（通过中间件可以办到）

二、分库分表

分表

分表分库一般分垂直和水平，垂直是指感觉业务来进行库的拆分，比如专门的用户库或者订单库这样子，但垂直还是无法解决单表数据量过大导致的性能会差的问题（这里可能会和上面矛盾，网上多数指的是 2000W 就会影响，但好像没有多少人谈过他们的表结构情况）。水平分指的是某一个表，里面的数据量非常大，我们按照一定的规则来进行一个拆分分流，比如把用户的数据由一直存放在用户表，变为可能这条数据是在用户1号表或者2号表这样子。规则可能是 范围（range）或者哈希（hash），也不知道我喜欢的取模算不算哈希。分了其实会带来一些问题的复杂性，比如分库那如何确保多库事务性的一致性、分布式锁等等很多，然后还有之前我们的 join 查询，现在可能就无法使用呢。

1、为什么要分表

当不使用分库分表的情况下，系统的性能瓶颈主要体现在：
当面临高并发场景的时候，为了避免Mysql崩溃（MySql性能一般的服务器建议2000/s读写并发以下），只能使用消息队列来削峰。
受制于单机限制。数据库磁盘容量吃紧。
数据库单表数据量太大，sql越跑越慢
而分库分表正是为了解决这些问题，提高数据库读写并发量，磁盘容量大大提高，单表数据量降低，提高查询效率。

2、垂直拆分和水平拆分

以表的维度来说：

垂直拆分 指根据表的字段进行拆分，其实很常见，有时候在数据库设计的时候就完成了，属于数据库设计范式，如订单表、订单支付表、商品表。
水平拆分 表结构一样，数据进行拆分。如原本的t_order表变为t_order_0,t_order_1，t_order_3

以库的维度来说：

垂直拆分 指把原本的大库，按业务不同拆到不同的库（微服务一般都是这么设计的，即专库专用）
水平拆分 一个服务对应多个库，每个库有相同的业务表，如库1有t_order表，库2也有t_order表。业务系统通过数据库中间件或中间层操作t_order表，分库操作对于业务代码透明。
所以，我们平常说的分库分表，一般都是指的水平拆分

分表分库的策略

1. hash分法，按一个键进行hash取模，然后分发到某张表或库。优点是可以平摊每张表的压力，缺点是扩容时会存在数据迁移问题。
2. range分法，按范围或时间分发，比如按某个键的值区间、或创建时间进行分发，优点是可以很方便的进行扩容，缺点是会造成数据热点问题。从分表上说还好，如果是分库，将导致某一个库节点压力过大，节点间负载不均。

php 分库分表hash算法

app/common.php 公共方法定义
$userid 也可以用下面的uid 发号器定义，通过哈希来就算出表名称

//哈希分表function get_hash_table($table, $userid) {$str = crc32($userid);if ($str < 0) {$hash = "0" . substr(abs($str), 0, 1);} else {$hash = substr($str, 0, 2);}return $table . "_" . $hash;}

控制器调用计算表名

public function index() {echo $table=get_hash_table('message', '18991').'<br>';echo $table=get_hash_table('message', '18993').'<br>';echo $table=get_hash_table('message', '18994').'<br>';
}

0、分表的方法（thinkphp）

public function getPartitionTableName($data=array()) {// 对数据表进行分区if(isset($data[$this->partition['field']])) {$field   =   $data[$this->partition['field']];switch($this->partition['type']) {case 'id':// 按照id范围分表$step    =   $this->partition['expr'];$seq    =   floor($field / $step)+1;break;case 'year':// 按照年份分表if(!is_numeric($field)) {$field   =   strtotime($field);}$seq    =   date('Y',$field)-$this->partition['expr']+1;break;case 'mod':// 按照id的模数分表$seq    =   ($field % $this->partition['num'])+1;break;case 'md5':// 按照md5的序列分表$seq    =   (ord(substr(md5($field),0,1)) % $this->partition['num'])+1;break;default :if(function_exists($this->partition['type'])) {// 支持指定函数哈希$fun    =   $this->partition['type'];$seq    =   (ord(substr($fun($field),0,1)) % $this->partition['num'])+1;}else{// 按照字段的首字母的值分表$seq    =   (ord($field{0}) % $this->partition['num'])+1;}}return $this->getTableName().'_'.$seq;}else{// 当设置的分表字段不在查询条件或者数据中// 进行联合查询，必须设定 partition['num']$tableName  =   array();for($i=0;$i<$this->partition['num'];$i++)$tableName[] = 'SELECT * FROM '.$this->getTableName().'_'.($i+1);$tableName = '( '.implode(" UNION ",$tableName).') AS '.$this->name;return $tableName;}
}

1、几种生成id的方式对比：

1、通过数据库自增
往公用的一张表（这张表是自增主键）插入一条数据，获取id的返回值，用这个id再去插入中间件当中去。oracle可以通过自增序列。
缺点：不适合并发高的场景，毕竟不管是自增序列还是采取自增键的方式来生成，会并发竞争写锁，效率太低。
2、UUID
缺点：uuid太长了，不规则
3、时间戳
一般联合其他业务字段拼接作为一个Id，如时间戳+用户id+业务含义编码
缺点：并发高容易重复
4、雪花算法 demo

<?php
class SnowFlake
{const TWEPOCH = 1288834974657; // 时间起始标记点，作为基准，一般取系统的最近时间（一旦确定不能变动）const WORKER_ID_BITS     = 5; // 机器标识位数const DATACENTER_ID_BITS = 5; // 数据中心标识位数const SEQUENCE_BITS      = 12; // 毫秒内自增位private $workerId; // 工作机器IDprivate $datacenterId; // 数据中心IDprivate $sequence; // 毫秒内序列private $maxWorkerId     = -1 ^ (-1 << self::WORKER_ID_BITS); // 机器ID最大值private $maxDatacenterId = -1 ^ (-1 << self::DATACENTER_ID_BITS); // 数据中心ID最大值private $workerIdShift      = self::SEQUENCE_BITS; // 机器ID偏左移位数private $datacenterIdShift  = self::SEQUENCE_BITS + self::WORKER_ID_BITS; // 数据中心ID左移位数private $timestampLeftShift = self::SEQUENCE_BITS + self::WORKER_ID_BITS + self::DATACENTER_ID_BITS; // 时间毫秒左移位数private $sequenceMask       = -1 ^ (-1 << self::SEQUENCE_BITS); // 生成序列的掩码private $lastTimestamp = -1; // 上次生产id时间戳public function __construct($workerId, $datacenterId, $sequence = 0){if ($workerId > $this->maxWorkerId || $workerId < 0) {throw new Exception("worker Id can't be greater than {$this->maxWorkerId} or less than 0");}if ($datacenterId > $this->maxDatacenterId || $datacenterId < 0) {throw new Exception("datacenter Id can't be greater than {$this->maxDatacenterId} or less than 0");}$this->workerId     = $workerId;$this->datacenterId = $datacenterId;$this->sequence     = $sequence;}public function nextId(){$timestamp = $this->timeGen();if ($timestamp < $this->lastTimestamp) {$diffTimestamp = bcsub($this->lastTimestamp, $timestamp);throw new Exception("Clock moved backwards.  Refusing to generate id for {$diffTimestamp} milliseconds");}if ($this->lastTimestamp == $timestamp) {$this->sequence = ($this->sequence + 1) & $this->sequenceMask;if (0 == $this->sequence) {$timestamp = $this->tilNextMillis($this->lastTimestamp);}} else {$this->sequence = 0;}$this->lastTimestamp = $timestamp;/*$gmpTimestamp    = gmp_init($this->leftShift(bcsub($timestamp, self::TWEPOCH), $this->timestampLeftShift));$gmpDatacenterId = gmp_init($this->leftShift($this->datacenterId, $this->datacenterIdShift));$gmpWorkerId     = gmp_init($this->leftShift($this->workerId, $this->workerIdShift));$gmpSequence     = gmp_init($this->sequence);return gmp_strval(gmp_or(gmp_or(gmp_or($gmpTimestamp, $gmpDatacenterId), $gmpWorkerId), $gmpSequence));*/return (($timestamp - self::TWEPOCH) << $this->timestampLeftShift) |($this->datacenterId << $this->datacenterIdShift) |($this->workerId << $this->workerIdShift) |$this->sequence;}protected function tilNextMillis($lastTimestamp){$timestamp = $this->timeGen();while ($timestamp <= $lastTimestamp) {$timestamp = $this->timeGen();}return $timestamp;}protected function timeGen(){return floor(microtime(true) * 1000);}// 左移 <<protected function leftShift($a, $b){return bcmul($a, bcpow(2, $b));}
}

5、uid 发号器 redis中间件生成
原理：利用redis单线程工作线程属性去维护一个自增变量。

2、ThinkPHP6 业务分表之一：UID 发号器

我们现在假设项目是新成立的，暂时没有一个技术债。目前我们要先规划一个用户表，打算划分 16 个表，按照取模的方式来查询。最先可能我们要考虑如何定义 UID 的问题，不过对我们 PHPer 来说不是什么难事，毕竟我们基本都不用 UUID 来做 UID 的，占用的空间会比较多，不利于索引。

但是不用 UUID ，选择了 INT 来做 UID，那我们要如何确保它的连续性和唯一性呢？业内常用的可能是雪花算法，但我选择自己写一个简易的发号器。

发号器需要加东西，然后取东西，我们可以利用 Redis List 来很好的实现我们需要的先进先出功能。通过一个命令或者说脚本，我们定时往列表中填上自增的 ID，然后在注册流程中来取最前面的。

代码层面
app/common.php 公共方法定义获取redis 的值

if(!function_exists('get_redis')) {function get_redis() {return new \Predis\Client('tcp://IP:端口', ['parameters' => ['password' => '密码',],]);}
}

定义一个发号器函数
app/command/GenerateUID.php

<?php
declare (strict_types = 1);namespace app\command;use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;class GenerateUID extends Command
{protected $redis;/** 发号器列表 键** @var string*/protected $cache_key = 'generate:uid';/** 锁 键** @var string*/protected $lock_key = 'generate:uid_lock';/** 发号器列表最大容量，默认为 500000** @var int*/protected $max_capacity = 5E5;public function __construct(){$this->redis = get_redis();parent::__construct();}protected function configure(){// 指令配置$this->setName('generate:uid')->setDescription('生成 UID');}protected function execute(Input $input, Output $output){$current_capacity = $this->get_list_length();// 计算发号器需要增加的数量$append_capacity = $this->max_capacity - $current_capacity;$uid_max = $this->get_uid_max();$output->writeln('当前最大UID：' . $uid_max);$output->writeln('当前剩余容量：' . $current_capacity);$output->writeln('最大存储容量：' . $this->max_capacity);$output->writeln('需要追加容量：' . $append_capacity);// 如果不需要增加则结束if($append_capacity === 0) {return;}$data = [];for($i = 1; $i <= $append_capacity; $i++) {$data[] = $uid_max + $i;}// 把需要加的数据进行分块，方便快速追加$data = array_chunk($data, 1000);// 加锁$lock = $this->redis->executeRaw(['SET',$this->lock_key,1,'EX',10 * 60,'NX',]);if($lock !== 'OK') {$output->writeln('获取锁失败');return;}try {foreach ($data as $item) {$this->redis->rpush($this->cache_key, $item);}} catch (\Exception $e) {$output->error($e->getMessage());} finally {// 释放锁$this->redis->del($this->lock_key);}}/** 获取发号器列表的长度** @return int*/protected function get_list_length() :int{return $this->redis->llen($this->cache_key);}/** 获取当前最大的 UID** @return int*/protected function get_uid_max() :int{$value = (int) $this->redis->lindex($this->cache_key, -1);if($value) {return $value;}return 0;}
}

config/console.php 定义一个打印函数类

<?php
return [// 指令定义'commands' => [...'generate:uid' => 'app\command\GenerateUID',],
];

执行命令

$ php think generate:uid
当前最大UID：0
当前剩余容量：0
最大存储容量：500000
需要追加容量：500000

3、ThinkPHP6 业务分表之二：用户

上一篇我们将了 UID 的发号器，那这一篇我们将要去实现用户的注册入库和简单的查询。

首先我们要实现数据表的识别。获取表名称
app/common.php

if(!function_exists('table_name')) {/** 获取表名称** @param string $name 表名* @param int|null $uid UID* @return string*/function table_name(string $name, ?int $uid = null) {$support_table = ['users' => 16, // 表名 => 分表数];if(!isset($support_table[$name])) {return $name;}// 如果没有传递 UID，那则需要调用其他方法来获取 UIDif(is_null($uid)) {$uid = (int) 1;}if((int) $uid === 0) {throw new \Exception('UID 值异常');}// 取 UID 和 分表数的模，值转化为小写十六进制return sprintf('%s_%x', $name, $uid % $support_table[$name]);}
}

执行写入操作
app/controller/Auth.php

<?php
namespace app\controller;use app\BaseController;
use think\facade\Db;
use think\helper\Str;class Auth extends BaseController
{/** 注册接口*/public function register(){$redis = get_redis();// 获取最左的值$uid = $redis->lpop('generate:uid');// 为空说明列表已经没内容了if(is_null($uid)) {return '无法获取 UID';}// 获取表名$table_name = table_name('users', $uid);// 插入数据Db::table($table_name)->insert(['id' => $uid,'nickname' => '随机生成' . Str::random(),]);return '注册成功';}
}

简单的分表写入

	
// 用于查询
$data = Db::table('sslq_question_1')->save(['keywords'=>1111]);

代码示例

第一次用数据表创建一个自增id 表，将id 拿出来放在即将插入的数据库中，每次插入先读当前条数拿出id 作为当前添加数据的自增id，结果 跑着跑着就卡了。毕竟每次都要多操作两次数据库，现在改成用redis 创建一个自增id 作为当前添加的数据id 。各有利弊吧，通过数据表获取自增每次执行插入需要查询总数量在插入一条自增数据，多操作2步数据库，用redis 直接读写，效率好，服务器也没有压力。（不知道reids 服务器挂机了，redis 会不会丢，丢的话，只能在写一个redis 备份）

分表写入

控制器

  /***添加试题**/public function add(){// $data = $this->postData();$model = new QuestionModel;for ($i=0; $i <10000; $i++) { $data = ['keywords'=>$i,'option'  =>'option','answer'  =>'answer','type'    =>rand(0,9999),'content_path'  =>'content_path','admin_user_id'  =>rand(0,9999)];$model->add($data);// try {//     //获取用户信息//     if ($model->add($data)) {//         // return $this->renderSuccess('添加成功');//     }//     return $this->renderError($model->getError() ?:'添加失败');// } catch (Exception $e) {//     throw new Exception($e->getMessage());// }}}

模型

	use think\cache\driver\Redis;    //需要使用到的类use think\facade\Cache;use think\facade\Config;/*** Created by* User: runtu* Date: 2024/9/23* Brief: 插入分表数据* docs: 模型初始化，获取表名称。插入到对应的表里面*/public function add($data){$redis = new Redis(Config::get('cache.stores.redis'));$uuid =  $redis->get('uuid')+1;$uuid = $uuid>0?$uuid:1;$redis->set('uuid',$uuid);$data['question_id'] = $uuid;Db::name($this->getTableName($uuid))->insert($data);//通过自建id表通过数据表自增来获取uuid// $uuid =  Db::name('question_id')->count()+1;// $data['question_id'] = $uuid;// Db::name($this->getTableName($uuid))->insert($data);// //通过自增函数来计算插入的id// return Db::name('question_id')->insert(['name'=>1]);}

数据库数据效果：

其他杂项

1、1亿条数据在PHP中实现Mysql数据库分表100张

当数据量猛增的时候，大家都会选择库表散列等等方式去优化数据读写速度。笔者做了一个简单的尝试，1亿条数据，分100张表。具体实现过程如下：
首先创建100张表：

$i=0;
while($i<=99){
echo "$newNumber \r\n";
$sql="CREATE TABLE `code_".$i."` (`full_code` char(10) NOT NULL,`create_time` int(10) unsigned NOT NULL,PRIMARY KEY  (`full_code`),
) ENGINE=MyISAM DEFAULT CHARSET=utf8";
mysql_query($sql);
$i++;

下面说一下我的分表规则，full_code作为主键，我们对full_code做hash
函数如下：

$table_name=get_hash_table('code',$full_code);
function get_hash_table($table,$code,$s=100){
$hash = sprintf("%u", crc32($code));
echo $hash;
$hash1 = intval(fmod($hash, $s));return $table."_".$hash1;
}

这样插入数据前通过get_hash_table获取数据存放的表名。
最后我们使用merge存储引擎来实现一张完整的code表

1 CREATE TABLE IF NOT EXISTS code (
2 full_code char(10) NOT NULL,
3 create_time int(10) unsigned NOT NULL,
4 INDEX(full_code)
5 ) TYPE=MERGE UNION=(code_0,code_1,code_2…) INSERT_METHOD=LAST ;

这样我们通过select * from code就可以得到所有的full_code数据了。

2、tp6 Sharding-Proxy企业分库分表最佳实践

tp6 Sharding-Proxy企业分库分表最佳实践