事務概念
參考： http://redis.cn/topics/transactions.html
事務是一個單獨的隔離操作：事務中的所有命令都會序列化、按順序地執行。事務在執行的過程中，不會被其他客戶端發送來的命令請求所打斷。
事務是一個原子操作：事務中的命令要么全部被執行，要么全部都不執行。
redis事務是一組命令的集合。多組命令進入到等待執行的事務隊列中，執行exec命令告訴redis將等待執行的事務隊列中的所有命令，按順序執行，返回值就是這些命令組成的列表。
Redis 事務可以一次執行多個命令， 具有下列保證：
- 批量操作在發送 EXEC 命令前被放入隊列緩存。
- 收到 EXEC 命令后進入事務執行，事務中任意命令執行失敗，其余的命令依然被執行。
- 在事務執行過程，其他客戶端提交的命令請求不會插入到事務執行命令序列中。
一個事務從開始到執行會經歷以下三個階段：
- 開始事務。
- 命令入隊。
- 執行事務。
事務中的錯誤：
- 事務在執行 EXEC 之前，入隊的命令可能會出錯。比如說，命令可能會產生語法錯誤（參數數量錯誤，參數名錯誤，等等），或者其他更嚴重的錯誤，比如內存不足（如果服務器使用 maxmemory 設置了最大內存限制的話）。
- 命令可能在 EXEC 調用之后失敗。舉個例子，事務中的命令可能處理了錯誤類型的鍵，比如將列表命令用在了字符串鍵上面，諸如此類。
從 Redis 2.6.5 開始，服務器會對命令入隊失敗的情況進行記錄，並在客戶端調用 EXEC 命令時，拒絕執行並自動放棄這個事務
在 EXEC 命令執行之后所產生的錯誤， 並沒有對它們進行特別處理： 即使事務中有某個/某些命令在執行時產生了錯誤， 事務中的其他命令仍然會繼續執行
如：
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set a 3
QUEUED
127.0.0.1:6379> lpop a
QUEUED
127.0.0.1:6379> exec
1) OK
2) (error) WRONGTYPE Operation against a key holding the wrong kind of value
redis 事務入隊只會檢查語法錯誤，對於exec后執行錯誤，沒有回滾措施。而且在事務中無法在客戶端做查詢判斷，只會得到queued，無法進行業務數據判斷，也是很坑。
原子性
一個事務是一個不可分割的最小工作單位,要么都成功要么都失敗。
原子操作是指你的一個業務邏輯必須是不可拆分的.比如你給別人轉錢,你的賬號扣錢,別人的賬號增加錢。
單個 Redis 命令的執行是原子性的，但 Redis 沒有在事務上增加任何維持原子性的機制，所以 Redis 事務的執行並不是原子性的。
看看下面幾個原子性的命令：
- HINCRBY key field increment 為哈希表 key 中的域 field 的值加上增量 increment
增量也可以為負數，相當於對給定域進行減法操作。
如果 key 不存在，一個新的哈希表被創建並執行 HINCRBY 命令。
如果域 field 不存在，那么在執行命令前，域的值被初始化為 0 - SETNX key value 只在鍵 key 不存在的情況下， 將鍵 key 的值設置為 value；若鍵 key 已經存在， 則 SETNX 命令不做任何動作。
事務命令
包含5個命令 MULTI、EXEC、DISCARD、WATCH、UNWATCH。
DISCARD 取消事務，放棄執行事務塊內的所有命令。
EXEC 執行所有事務塊內的命令。
MULTI 標記一個事務塊的開始。
UNWATCH 取消 WATCH 命令對所有 key 的監視。
WATCH key [key ...] 監視一個(或多個) key ，如果在事務執行之前這個(或這些) key 被其他命令所改動，那么事務將被打斷。
樂觀鎖
樂觀的認為數據不會出現沖突，使用version或timestamp來記錄判斷。樂觀鎖的優點開銷小，不會出現鎖沖突。
可利用watch命令監聽key，實現樂觀鎖，來保證不會出現沖突，應用場景比如秒殺來防止超賣。
秒殺偽代碼如下：
WATCH 鎖定量
MULTI
incr 鎖定量
if 鎖定量 <= 庫存量
減庫存
EXEC
悲觀鎖
了解下相關命令
- SETNX（SET if Not eXists） key value 只在鍵 key 不存在的情況下， 將鍵 key 的值設置為 value，返回值：命令在設置成功時返回 1 ， 設置失敗時返回 0
- INCR KEY 為鍵 key 儲存的數字值加上一。
如果鍵 key 不存在， 那么它的值會先被初始化為 0 ， 然后再執行 INCR 命令。
如果鍵 key 儲存的值不能被解釋為數字， 那么 INCR 命令將返回一個錯誤。
命令會返回鍵 key 在執行加一操作之后的值 - SET key value [EX seconds] [PX milliseconds] [NX|XX] NX等同於SETNX操作，EX seconds 將鍵的過期時間設置為 seconds 秒
了解下搶購模擬代碼：
<?php
namespace app\controllers;
use Yii;
use yii\web\Controller;
use app\modules\Common;
/**
* 模擬搶購處理
* Class ShopController
* @package app\controllers
*/
class ShopController extends Controller
{
public $goods = 'huawei P20';
//初始化數據
public function actionInit(){
$redis = Yii::$app->redis;
$redis->set('goodNums',100); //設置庫存
$redis->del('order'); //清空搶購訂單
die('success');
}
//悲觀鎖
//setnx 實現，有個問題 expire失敗（1.人為錯誤;2.redis崩了）了，這個鎖就持久化，一直被鎖了
public function actionBuy(){
$userId = mt_rand(1,99999999);
$goods = $this->goods;
$redis = Yii::$app->redis;
$lock = $goods;
try {
$inventory['num'] = $redis->get('goodNums');
if($inventory['num']<=0){
self::removeLock($lock);
throw new \Exception('活動結束');
}
if( $redis->setnx($lock,1) ){
$redis->expire($lock,60);//設置過期時間，防止死鎖
//業務處理 減庫存，創建訂單
$redis->decr('goodNums');
$redis->sadd('order',$userId);
//todo 實際業務處理時間不可控，所以需要調整過期時間，在業務處理完進行剩余生命時間的判斷，沒找到回滾業務
$this->removeLock($lock);
}else{
throw new \Exception($userId.' 搶購失敗');
}
Common::addLog('shop.log',$userId.' 搶購成功');
}catch (\Exception $e){
$this->removeLock($lock);
Common::addLog('shop.log',$e->getMessage());
}
die('success');
}
//刪除鎖
protected function removeLock( $lock ){
$redis = Yii::$app->redis;
return $redis->del($lock);
}
//悲觀鎖
//incr 解決expire失效，解鎖
public function actionBuy2(){
$userId = mt_rand(1,99999999);
$goods = $this->goods;
$redis = Yii::$app->redis;
$lock = $goods;
try {
$inventory['num'] = $redis->get('goodNums');
if($inventory['num']<=0){
$this->removeLock($lock);
throw new \Exception('活動結束');
}
$lockset = $redis->incr($lock);
if( !$lockset ){
throw new \Exception($userId.' 搶購失敗');
}
if($lockset==1){
$redis->expire($lock,60);//設置過期時間，防止死鎖
//業務處理 減庫存，創建訂單
$redis->decr('goodNums');
$redis->sadd('order',$userId);
$this->removeLock($lock);
}
//鎖的數量大於1並且沒有設置過期時間，失敗處理
if( $lockset>1 && $redis->ttl($lock)===-1 ){
$this->removeLock($lock);
throw new \Exception($userId.' 搶購失敗');
}
Common::addLog('shop.log',$userId.' 搶購成功');
}catch (\Exception $e){
$this->removeLock($lock);
Common::addLog('shop.log',$e->getMessage());
}
die('success');
}
//悲觀鎖
//set key value [expiration EX seconds|PX milliseconds] [NX|XX] 原子命令（redis必須大於2.6版本）
public function actionBuy3(){
$userId = mt_rand(1,99999999);
$goods = $this->goods;
$redis = Yii::$app->redis;
$lock = $goods;
try {
$inventory['num'] = $redis->get('goodNums');
if($inventory['num']<=0){
$this->removeLock($lock);
throw new \Exception('活動結束');
}
$lockset = $redis->set($lock,1,'EX',60,'NX');
if( !$lockset ){
throw new \Exception($userId.' 搶購失敗');
}
if($lockset==1){
//業務處理 減庫存，創建訂單
$redis->decr('goodNums');
$redis->sadd('order',$userId);
$this->removeLock($lock);
}
Common::addLog('shop.log',$userId.' 搶購成功');
}catch (\Exception $e){
$this->removeLock($lock);
Common::addLog('shop.log',$e->getMessage());
}
die('success');
}
# 樂觀鎖
public function actionBuy4(){
$userId = mt_rand(1,99999999);
$goods = $this->goods;
$redis = Yii::$app->redis;
$lock = $goods;
try {
$inventory['num'] = $redis->get('goodNums');
if($inventory['num']<=0){
throw new \Exception('活動結束');
}
$redis->watch($lock);
$redis->multi();
//todo：這里還需要重新判斷下庫存，否則會出現超發，高並發情況下$inventory['num']肯定會出現同時讀取一個值；為了方便測試，沒寫db操作
//redis事務是將命令放入隊列中，無法取goodNums來判斷庫存是否結束，此處使用數據庫來判斷庫存合理
//業務處理 減庫存，創建訂單
$redis->decr('goodNums');
$redis->sadd('order',$userId);
$redis->exec();
Common::addLog('shop.log',$userId.' 搶購成功');
}catch (\Exception $e){
$redis->discard();
Common::addLog('shop.log',$e->getMessage());
}
die('success');
}
# 隊列實現，不做詳述
}
並發控制及過期時間
服務器訪問並發比較大，無效訪問頻繁，比如說頻繁請求接口，爬蟲頻繁訪問服務器，搶購瞬時請求過大，我們需要限流處理。
限流：對訪問來源計數，超過設定次數，設置過期時間，提醒訪問頻繁，稍后再試
limits=500 #設置1秒內限制次數50
if EXISTS userid
return '訪問頻繁，鎖定時間剩余（ttl userid）秒'
if userid_count_time > limits
exprice userid,3600
return '訪問頻繁，稍后再試'
else
MUlTI
incr userid_count_time # 對用戶每秒的請求進行原子遞增計數
exprice userid_count_time , 60
EXEC
//使用事務的目的是避免執行錯誤中斷，userid_count_time持久化到磁盤，高並發下這個很有必要
計數器限流，缺點也很大，可能會超過限制數。相比下，高並發 漏桶算法、令牌桶算法更適合做限流，此處不做深究。
隊列
可以用隊列來做異步任務處理，實現解耦過程。來看看以下可能在使用中需要注意的問題
隊列防丟失
運用數據格式list，lpush、rpop就可以入隊、出隊，但是會有個問題 假設出隊的業務執行發生錯誤，數據會不會因此丟失，所以需要確保出隊時確實被消費了，可以參考下面偽代碼處理：
while(val = lrange(list,0,-1))
try{
//對val這條數據的業務代碼處理
rpop(list)
}catch(Exception e){
//記錄錯誤,通知programmer處理
break;
}
參考下lrange語法
阻塞隊列
使用隊列還有另外一個問題，空隊列會一直占用redis連接。利用redis隊列操作的阻塞命令我們可以解決這個問題。
來看下這個命令：
BRPOP key [key …] timeout
它是 RPOP key 命令的阻塞版本，當給定列表內沒有任何元素可供彈出的時候，連接將被 BRPOP 命令阻塞，直到等待超時或發現可彈出元素為止。
阻塞執行意味着不會一直占用連接，如果這個隊列一直是空的，那么當客戶端連接超過redis最大連接時間就會自動斷掉，代碼可能需要做些redis重連操作，但是這個過程相對於一直占用連接，頻繁io請求，
所帶來的redis性能影響將減少很多。
偽代碼：
try(
while ( var = BRPOP key ){
//對var業務處理過程
}
catch(Exception e){
//redis連接中斷 重連操作
}
時間區間控制
雖然阻塞隊列能解決這個連接占用，但是日常的任務我們基本上借助cron執行，這也就是個定時執行的概念，假設采取上面的方法，勢必存在多個進程同時再跑，長此以往系統的進程數就占滿了，怎么辦呢？
使用時間區間來限制進程運行時間,假設腳本每10分鍾執行一次，執行9分鍾中斷：
偽代碼
time = now()+60*9
while ( now()< time ) {
var = RPOP key
if( var ){
//對var業務處理過程
}else{
sleep(2) //空數據，休息2秒 避免頻繁連接，占用redis資源
}
}
也可以加上數據防丟失的處理，具體根據業務需求靈活處理。
持久化
服務器中的非空數據庫以及數據庫中的健值對統稱數據庫狀態。
redis是內存數據庫，數據庫狀態存在內存中，一旦服務器崩掉，服務器狀態就會消失不見，所以需要將數據庫狀態存與磁盤文件中。
RDB
定期的將數據庫狀態保存在一個RDB快照文件中，RDB文件是一個經過壓縮的二進制文件，通過該文件可還原生成RDB文件時的數據庫狀態。
觸發方式：手動和自動
RDB 文件的創建和載入
redis命令：SAVE、BGSAVE
SAVE會阻塞Redis服務器進程，直到RDB文件創建完畢為止，在服務器進程阻塞期間，服務器不能處理任何命令請求。
BGSAVE命令會派生出一個子進程，然后由子進程負責創建RDB文件，服務器進程（父進程）繼續處理命令請求。
自動觸發
redis.conf 中配置
save 900 1 # 表示900 秒內如果至少有 1 個 key 的值變化，則保存
save 300 10 # 表示300 秒內如果至少有 10 個 key 的值變化，則保存
save 60 10000 # 表示60 秒內如果至少有 10000 個 key 的值變化，則保存
“save m n”。表示m秒內數據集存在n次修改時，自動觸發BGSAVE。
偽代碼
def SAVE():
#創建RDB文件
rdbSave()
def BGSAVE():
#創建子進程
pid = fork()
if pid == 0:
#子進程負責創建RDB文件
rdbSave()
#完成之后向父進程發送信號
signal_parent()
elif pid ＞ 0:
#父進程繼續處理命令請求，並通過輪詢等待子進程的信號
handle_request_and_wait_signal()
else:
#處理出錯情況
handle_fork_error()
AOF
AOF持久化功能實現分為命令追加（append）、文件寫入（wirte）、文件同步（sync）三個步驟。
每一個寫命令都通過write函數追加到 appendonly.aof 中,配置方式：啟動 AOF 持久化的方式
偽代碼
def eventLoop():
while True:
#處理文件事件，接收命令請求以及發送命令回復
#處理命令請求時可能會有新內容被追加到 aof_buf緩沖區中
processFileEvents()
#處理時間事件
processTimeEvents()
#考慮是否要將 aof_buf中的內容寫入和保存到 AOF文件里面
flushAppendOnlyFile()
命令追加
服務器在執行一個寫命令之后，會以協議格式將執行的寫命令追加到服務器狀態的aof_buf緩沖區的末尾。
文件寫入、同步
操作系統中，用戶調用write函數寫入，將一些數據寫入到文件時，為了提高存儲的效率，操作系統通常會將數據暫時保存在一個內存緩沖區里面，緩沖區滿了或者超過指定時間，真正將緩沖區數據存儲到磁盤，提高了效率，但是如果停機，也會造成緩沖區內的數據丟失，
系統提供了fsync、fdatasync兩個同步函數，會強制讓操作系統立即將緩沖區的數據寫入硬盤，確保數據的安全性。
AOF持久化配置 redis.conf ：
appendonly yes #開啟AOF
appendfilename "appendonly.aof" #默認存儲路徑
# appendfsync 設置持久化策略，三種：
#appendfsync always # 每次有數據修改發生時AOF緩沖區數據都會寫入AOF文件並同步 （效率最慢但安全性最高）
appendfsync everysec # 每秒鍾寫入AOF文件並同步一次，該策略為AOF的缺省策略。（效率高，即便丟失數據只會丟失1秒的數據）
#appendfsync no # 緩沖區的內容寫入到AOF文件，但並不會對AOF文件進行同步，何時同步由操作系統來決定（效率高，丟失上一次同步到這一次的全部AOF數據）
appendonly yes開啟 AOF 之后，Redis 每執行一個修改數據的命令，都會把它添加到 AOF 文件中，當 Redis 重啟時，將會讀取 AOF 文件進行“重放”以恢復到 Redis 關閉前的最后時刻。
RDB、AOF優缺點
RDB優缺
AOF優缺
使用 AOF 持久化會讓 Redis 變得非常耐久（much more durable）：你可以設置不同的 fsync 策略，比如無 fsync ，每秒鍾一次 fsync ，或者每次執行寫入命令時 fsync 。 AOF 的默認策略為每秒鍾 fsync 一次，在這種配置下，
Redis 仍然可以保持良好的性能，並且就算發生故障停機，也最多只會丟失一秒鍾的數據（ fsync 會在后台線程執行，所以主線程可以繼續努力地處理命令請求）。
對於相同的數據集來說，AOF 文件的體積通常要大於 RDB 文件的體積。根據所使用的 fsync 策略，AOF 的速度可能會慢於 RDB。 在一般情況下， 每秒 fsync 的性能依然非常高， 而關閉 fsync 可以讓 AOF 的速度和 RDB 一樣快，
即使在高負荷之下也是如此。 不過在處理巨大的寫入載入時，RDB 可以提供更有保證的最大延遲時間（latency）。
隨着服務器時間的流逝，AOF文件的體積會越來越大。
排序
redis可以當作數據庫來存貯數據，如何解決排序查詢呢？
SORT命令：
redis禁用危險命令
keys *
雖然其模糊匹配功能使用非常方便也很強大，在小數據量情況下使用沒什么問題，數據量大會導致 Redis 鎖住及 CPU 飆升，在生產環境建議禁用或者重命名！
flushdb
刪除 Redis 中當前所在數據庫中的所有記錄，並且此命令從不會執行失敗
flushall
刪除 Redis 中所有數據庫中的所有記錄，不只是當前所在數據庫，並且此命令從不會執行失敗。
config
客戶端可修改 Redis 配置。
參考：
https://blog.csdn.net/a169388842/article/details/82838818
redis配置
# 綁定ip，指定地址域連接
bind 192.168.1.100 10.0.0.1