第淺談并發(fā)處理PHP進(jìn)程間通信之外部介質(zhì)目錄進(jìn)程間通信文件flock代碼實(shí)現(xiàn)mysqlselectforupdate代碼實(shí)現(xiàn)redisincrSETNX代碼實(shí)現(xiàn)優(yōu)化總結(jié)

進(jìn)程間通信

進(jìn)程間通信，指至少兩個(gè)進(jìn)程或線(xiàn)程間傳送數(shù)據(jù)或信號(hào)的一些技術(shù)或方法。進(jìn)程是計(jì)算機(jī)系統(tǒng)分配資源的最小單位(嚴(yán)格說(shuō)來(lái)是線(xiàn)程)。每個(gè)進(jìn)程都有自己的一部分獨(dú)立的系統(tǒng)資源，彼此是隔離的。為了能使不同的進(jìn)程互相訪(fǎng)問(wèn)資源并進(jìn)行協(xié)調(diào)工作，才有了進(jìn)程間通信。

根據(jù)定義可知，要進(jìn)行進(jìn)程間通信，我們需要解決兩個(gè)問(wèn)題：

互相訪(fǎng)問(wèn)：消息傳輸和暫時(shí)存儲(chǔ)介質(zhì)選擇問(wèn)題；

協(xié)調(diào)工作：消息的存取沖突問(wèn)題；

文章介紹的中心就是圍繞著這么兩點(diǎn)來(lái)說(shuō)的，為了更使文章更簡(jiǎn)明，這邊以之前在公司做的一個(gè)需求為例:

需要一個(gè)循環(huán)ID生成器，循環(huán)生成從Min到Max的數(shù)字ID，在ID遞增到Max后，返回到Min重新開(kāi)始遞增；必須能保證多個(gè)進(jìn)程并發(fā)請(qǐng)求時(shí)生成的ID不同。

此需求要解決的問(wèn)題恰好為我們要解決的進(jìn)程間通信需要解決的兩個(gè)問(wèn)題：

需要一個(gè)消息傳輸通道來(lái)傳輸和存儲(chǔ)當(dāng)前的遞增值。這個(gè)比較容易解決，我們常用的文件、數(shù)據(jù)庫(kù)、session、緩存等都能做到。

需要解決多進(jìn)程同時(shí)訪(fǎng)問(wèn)生成器生成相同ID的問(wèn)題。要滿(mǎn)足這個(gè)需要就必須要用到鎖了，而且為了保證多個(gè)進(jìn)程讀取的數(shù)據(jù)是不同的，需要互斥鎖，另外為了能保證調(diào)用成功率，鎖的獲取最好能實(shí)現(xiàn)自旋。

本文通過(guò)此需求的不同實(shí)現(xiàn)，來(lái)介紹通過(guò)外部介質(zhì)進(jìn)行的進(jìn)程間通信的方式。另外，不只PHP語(yǔ)言，其他語(yǔ)言也能使用這些方法。

文件

flock

文件是最基本的存儲(chǔ)介質(zhì)，它當(dāng)然可以作為消息的傳輸通道來(lái)使用。文件的存取各種語(yǔ)言都有各自的多種方案，問(wèn)題點(diǎn)是多進(jìn)程并發(fā)時(shí)的沖突問(wèn)題。

解決存取沖突問(wèn)題我們使用PHP的flock()函數(shù)：

boolflock(resource$handle,int$operation[,int$wouldblock])

$handler是使用fopen($path_to_file)獲取到的文件句柄；

$operation是對(duì)文件加鎖的方式，有以下值可選：

LOCK_SH(獲取共享鎖)/LOCK_EX(獲取互斥鎖)/LOCK_UN(解鎖)

這里我們選用互斥鎖，一個(gè)進(jìn)程獲取到互斥鎖后，其他進(jìn)程在嘗試獲取鎖會(huì)被阻塞，直到鎖被釋放，即實(shí)現(xiàn)了自旋；

此外，還有一個(gè)參數(shù)LOCK_NB，flock在獲取不到鎖時(shí)，默認(rèn)會(huì)阻塞住直到鎖被其他進(jìn)程釋放，傳入LOCK_NB與LOCK_SH或LOCK_EX進(jìn)行或運(yùn)算結(jié)果（LOCK_EX|LOCK_NB），flock在鎖被其他進(jìn)程占有時(shí)，不會(huì)阻塞，而是直接返回false，這里僅作介紹，我們并不使用它。

$wouldblock參數(shù)是一個(gè)引用值，在獲取不到鎖，且不阻塞模式時(shí)，$wouldblock會(huì)被設(shè)置為true；（手冊(cè)中說(shuō)阻塞時(shí)才會(huì)被設(shè)置為true。其實(shí)我也奇怪這個(gè)變量名的。不知道是不是bug，我的PHP版本是5.4.5，有知道的煩請(qǐng)解惑）

代碼實(shí)現(xiàn)

下面是循環(huán)ID生成器代碼，說(shuō)明在注釋中:

functiongetCycleIdFromFile($max,$min=0){

$handler=fopen('/tmp/cycle_id_generator.txt','c+');

if(!flock($handler,LOCK_EX)){

thrownewException('error_get_file_lock!');

$cycle_id=trim(fread($handler,9));

$cycle_id++;

if($cycle_id$max){

$cycle_id=$min;

//文件指針?lè)祷氐轿募^,并向文件內(nèi)寫(xiě)入新的cycle_id

rewind($handler);

fwrite($handler,$cycle_id);

//多寫(xiě)入一些空格為了防止數(shù)值升到多位后,突然置為少位后面的數(shù)字仍保留

fwrite($handler,str_repeat('',9));

flock($handler,LOCK_UN);

return$cycle_id;

}

mysql

selectforupdate

我們常用的mysql也可以被當(dāng)作中間介質(zhì)來(lái)實(shí)現(xiàn)進(jìn)程間的通信，我們規(guī)定好某一個(gè)數(shù)據(jù)表內(nèi)的某一行數(shù)據(jù)作為消息交換的中轉(zhuǎn)站，使用mysql自帶的鎖來(lái)協(xié)調(diào)多個(gè)進(jìn)程的存取沖突。

事務(wù)的設(shè)計(jì)目的就是為了解決多進(jìn)程并發(fā)查詢(xún)時(shí)數(shù)據(jù)沖突的問(wèn)題，可是我們常用的事務(wù)只能保證數(shù)據(jù)沖突時(shí)會(huì)被回滾，數(shù)據(jù)不會(huì)出現(xiàn)錯(cuò)誤，并不能實(shí)現(xiàn)請(qǐng)求的并行化。對(duì)一些數(shù)據(jù)沖突回滾的請(qǐng)求，需要我們?cè)谕鈱犹砑舆壿嬛卦嚒?/p>

這里介紹mysql的一種語(yǔ)法：selectforupdate，會(huì)給固定數(shù)據(jù)加上互斥鎖，且另一個(gè)請(qǐng)求在獲取鎖失敗時(shí)，會(huì)阻塞至獲取鎖成功，mysql幫我們實(shí)現(xiàn)了自旋；

用法如下：

1.關(guān)閉mysql的自動(dòng)提交，自動(dòng)提交默認(rèn)打開(kāi)，除非使用transition語(yǔ)句顯示開(kāi)啟事務(wù)，默認(rèn)會(huì)將每一條sql作為一個(gè)事務(wù)直接提交執(zhí)行，這里關(guān)閉。setautocommit=0;

2.使用selectforupdate語(yǔ)句給數(shù)據(jù)添加互斥鎖。注意：需求mysql的innodb引擎支持;

3.進(jìn)行數(shù)據(jù)更新和處理操作；

4.主動(dòng)提交事務(wù)，并將自動(dòng)提交恢復(fù)；commit;setautocommit=1;

然后是代碼實(shí)現(xiàn)：

//數(shù)據(jù)庫(kù)連接實(shí)現(xiàn)各有不同，demo可以自己修改一下。

functiongetCycleIdFromMysql($max,$min=0){

Db::db()-execute('setautocommit=0');

$res=Db::db()-qsqlone('SELECTcycle_idFROMcycle_id_generatorWHEREid=1FORUPDATE');

$cycle_id=$res['cycle_id']+1;

if($cycle_id$max){

$cycle_id=$min;

Db::db()-execute("UPDATEcycle_id_generatorSETcycle_id={$cycle_id}WHEREid=1");

Db::db()-execute('commit');

Db::db()-execute('setautocommit=1');

return$cycle_id;

}

redis

incr

redis是我們常用的緩存服務(wù)器，由于其使用內(nèi)存存儲(chǔ)數(shù)據(jù)，性能很高。我們使用一個(gè)固定的普通鍵來(lái)作為消息中轉(zhuǎn)站，然后利用其incr命令的原子性和其執(zhí)行結(jié)果（遞增后的值），實(shí)現(xiàn)cycle_id的遞增。

incr(key)若key不存在，redis會(huì)先將值設(shè)置為0，然后執(zhí)行遞增操作；

遞增沒(méi)有問(wèn)題，可是我們還有個(gè)需求是在要其值達(dá)到max時(shí)，再將其置為min，這時(shí)就可能會(huì)出現(xiàn)進(jìn)程A在更新值為min時(shí)，另一個(gè)進(jìn)程B也檢測(cè)到值大于了max，然后將值置為min，可是這時(shí)的值已經(jīng)不是max，即發(fā)生了值重復(fù)更新，那么返回的值必然會(huì)有重復(fù)；

這時(shí)，我們就需要自己來(lái)實(shí)現(xiàn)鎖了。

SETNX

redis的SETNX命令檢測(cè)某一個(gè)key是否存在，若不存在，則將key的值設(shè)置為value,并返回結(jié)果1；若key已存在，則設(shè)置失敗，返回值0。

SETNXkeyvalue

它能實(shí)現(xiàn)鎖是因?yàn)樗且粋€(gè)原子命令，即檢測(cè)key是否存在和設(shè)置key值在一個(gè)事務(wù)內(nèi)，不會(huì)出現(xiàn)同時(shí)兩個(gè)進(jìn)程都檢測(cè)到key不存在，然后同時(shí)去設(shè)置key的情況。

我們以另一個(gè)值的存在與否，來(lái)表示cycle_id是否正在被另一個(gè)進(jìn)程修改。

functiongetCycleIdFromRedis($max,$min=0){

$redis=newRedis();

$redis-connect('127.0.0.1',6379);

$key_id='cycle_id_generator';

$cycle_id=$redis-incr($key_id);

if($cycle_id$max){

//設(shè)置"鎖鍵"的結(jié)果=獲取互斥結(jié)果

$key_lock='cycle_id_lock';

if(!$redis-setnx($key_lock,1)){

returnnull;

$cycle_id=$min;

$redis-set($key_id,$cycle_id);

//最后別忘記釋放互斥鎖

$redis-delete($key_lock);

$redis-close();

return$cycle_id;

}

注意：由于redis里沒(méi)有能實(shí)現(xiàn)自旋鎖的命令，如果需求最高的獲取成功率，我們?cè)跈z測(cè)到cycle_id已經(jīng)是最大值，且試圖修改獲取鎖失敗時(shí)，退出重試，在外層進(jìn)行重試。

functiongetCycleId($max,$min=0){

$cycle_id=getCycleIdFromRedis($max,$min);

if(!is_null($cycle_id)){

return$cycle_id;

//稍微等待下正在更改的進(jìn)程

usleep(500);

//這里使用遞歸,直至獲取成功并發(fā)很高,cycle_id重置很頻繁時(shí)慎用.

returngetCycleId($max,$min);

}

優(yōu)化

審查代碼我們會(huì)發(fā)現(xiàn)，如果max-min的值很小的話(huà)，redis會(huì)需要經(jīng)常重置key的值，也就經(jīng)常需要加鎖，重試也就很多。這里，我提供一個(gè)優(yōu)化方法：

我們將其max設(shè)置為一個(gè)很大的值（要能被max-min整除），返回值時(shí)稍做處理，返回$current%($max-$min)+$min;。這樣，key需要遞增到一個(gè)很大的值才會(huì)被重置，加鎖邏輯和外層邏輯會(huì)很少執(zhí)行到，達(dá)到提升效率的目的