发布时间：January 9, 2014 // 分类：PHP // No Comments
Yum安装的PHP需安装扩展包php-process：
<?php
$shm_key = intval(bin2hex('node'), '16');
$memsize = 120;
$shm_h = shm_attach($shm_key, $memsize, 0644);
$var_key = intval(bin2hex('key'), '16');
$var_value = 'hello';
shm_put_var($shm_h, $var_key, $var_value);
echo shm_get_var($shm_h, $var_key);
shm_remove_var($shm_h, $var_key);
shm_detach($shm_h);
?>发布时间：January 9, 2014 // 分类：PHP // No Comments
多进程：
<?php
//生成key
$message_queue_key = ftok(__FILE__, 'a');
//根据生成的key新建队列，也可自定，如123456
$message_queue = msg_get_queue($message_queue_key, 0666);
$pids = array();
for ($i = 0; $i < 5; $i++) {
//创建子进程
$pids[$i] = pcntl_fork();
if ($pids[$i]) {
echo "No.$i child process was created, the pid is $pids[$i]\r\n";
pcntl_wait($status);//非阻塞的线程等待,防止僵尸进程的出现
} elseif ($pids[$i] == 0) {
$pid = posix_getpid();
echo "process.$pid is writing now\r\n";
//写队列
msg_send($message_queue, 1, "this is process.$pid's data\r\n");
posix_kill($pid, SIGTERM);
}
}
do {
//读队列
msg_receive($message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT);
echo $message;
//获取队列内消息数
$a = msg_stat_queue($message_queue);
if($a['msg_qnum'] == 0){
break;
}
} while(true)
?>父子进程：
<?php
$message_queue_key = ftok(__FILE__, 'a');
$message_queue = msg_get_queue($message_queue_key, 0666);
$pid = pcntl_fork();
if ($pid==-1) {
die("cannot fork");
} else if ($pid) { //父进程
pcntl_wait($status);
msg_receive($message_queue, 0, $message_type, 1024, $message, true, MSG_IPC_NOWAIT);
echo $message;
} else {
$pid = posix_getpid(); //子进程
msg_send($message_queue, 1, "this is process.$pid's data\r\n");
}
?>发布时间：January 8, 2014 // 分类：NoSQL // No Comments
Tokyo Cabinet支持Table数据结构，类似于SQL，可按条件查找索引，记录下：
#新建数据库，实际上类似于SQL表
tctmgr create user.tct
#key1为主健，name和passwd为字段
tctmgr put user.tct key1 "name" "u1" "passwd" "123"
#列出当前表数据
tctmgr list -pv user.tct
#按字段查找，类似于select * from where name = u1 limit 15;
tctmgr search -pv -m 15 user.tct name STREQ "u1"
#建立索引
tctmgr setindex user.tct namePHP接口：
<?php
$tt = new TokyoTyrantTable("localhost", 1978);
$query = $tt->getQuery();
$query->addCond("name", TokyoTyrant::RDBQC_STREQ, "u1")->setLimit(5, 5);
var_dump($query->search());
?>发布时间：January 7, 2014 // 分类：消息队列 // No Comments
根据Tokyo Tyrant自带的Lua队列脚本修改，Kyoto Cabinet和Tokyo Tyrant的API有很大不同。
kt = __kyototycoon__
db = kt.db
-- 记录日志
if kt.thid == 0 then
kt.log("system", "the Lua script has been loaded")
end
-- 入队列
function enqueue(inmap, outmap)
local key = inmap.key
local value = inmap.value
--队列值自增1，空从0开始
local id = db:increment_double(key, 1)
if not id then
return kt.RVEINTERNAL
end
key = string.format("%s-%012d", key, id)
if not db:add(key, value) then
return kt.RVEINTERNAL
end
outmap[key] = "ok"
return kt.RVSUCCESS
end
-- 出队列
function dequeue(inmap, outmap)
local key = inmap.key
local max = inmap.max
max = tonumber(max)
if not max or max < 1 then
max = 1
end
key = string.format("%s-", key)
--匹配队列前缀，返回多个匹配的key
local keys = db:match_prefix(key, max)
local res = ""
for i = 1, #keys do
local key = keys[i]
local value = db:get(key)
if db:remove(key) and value then
--要返回的结果
outmap[keys[i]] = value
end
end
return kt.RVSUCCESS
end
-- 查看队列大小
function queuesize(inmap, outmap)
local key = inmap.key
key = string.format("%s-", key)
local keys = db:match_prefix(key)
outmap.size = #keys
return kt.RVSUCCESS
end
-- 重置队列ID从0开始
function queuereset(inmap, outmap)
local key = inmap.key
if not key then
return kt.RVEINVALID
end
if not db:remove(key) then
local err = db:error()
if err:code() == kt.Error.NOREC then
return kt.RVELOGIC
end
return kt.RVEINTERNAL
end
return kt.RVSUCCESS
end使用：
ktremotemgr script -host 192.168.1.3 -port 1978 enqueue key queue value value1
ktremotemgr script -host 192.168.1.3 -port 1978 dequeue key queue max 10
curl "http://192.168.1.3:1978/rpc/play_script?name=enqueue&_key=queue&_value=value1"
curl "http://192.168.1.3:1978/rpc/play_script?name=dequeue&_key=queue&_max=10"发布时间：January 7, 2014 // 分类：消息队列 // No Comments
Tokyo Tyrant官方提供的Lua脚本扩展：
-- 入队列
function enqueue(key, value)
--通过队列名为key记录入队列的尾部，自加1为当次入队列的起始ID
local id = _adddouble(key, 1)
if not id then
return nil
end
--队列key格式为队列名称\t加12位数字
key = string.format("%s\t%012d", key, id)
if not _putkeep(key, value) then
return nil
end
return "ok"
end
-- 出队列
function dequeue(key, max)
max = tonumber(max)
if not max or max < 1 then
max = 1
end
key = string.format("%s\t", key)
--匹配队列前缀，返回匹配的多个key
local keys = _fwmkeys(key, max)
local res = ""
for i = 1, #keys do
local key = keys[i]
local value = _get(key)
--删除要出队列的key并返回队列值
if _out(key) and value then
res = res .. value .. "\n"
end
end
return res
end
-- 查看队列大小
function queuesize(key)
key = string.format("%s\t", key)
local keys = _fwmkeys(key)
//队列数组大小
return #keys
endtcrmgr ext -port 1978 127.0.0.1 enqueue myqueue value1
tcrmgr ext -port 1978 127.0.0.1 dequeue myqueue 10通过TT入队列：
<?php
function dequeue(){
$tt = new TokyoTyrant("localhost", '1978');
$content = $tt -> ext('dequeue', '0', 'queue', 10);
$content = preg_split('/\n/', $content, -1, PREG_SPLIT_NO_EMPTY);
return $content;
}
//队列名称
$file = 'file.csv';
$queuename = 'queue';
$tt = new TokyoTyrant("localhost", '1978');
$content = file_get_contents($file);
$content = preg_split('/\n/', $content, -1, PREG_SPLIT_NO_EMPTY); //分割字符串为数组
//队列开始ID
//print_r($content);
foreach($content as $key => $value)
{
if ($key !== '')
{
$tmpvalue = explode(',',$value);
preg_match_all("/[0-9]{1}/", $tmpvalue[0], $arrNum, PREG_SET_ORDER);
if (count($arrNum) > 8)
{
$tt->ext('enqueue','0', $queuename, $value );
//echo $value."\n";
}
}
}
?>大数据通过TC PHP扩展入队列：
<?php
//队列名称
$file = 'file.txt';
$queuename = 'queue';
$db = new TCHDB();
$db->open('/var/queue/db.tch', TCHDB::OWRITER | TCHDB::OCREAT);
$content = file_get_contents($file);
$content = preg_split('/\n/', $content, -1, PREG_SPLIT_NO_EMPTY); //分割字符串为数组
//队列开始ID
$id = 1;
//print_r($content);
foreach($content as $key => $value)
{
if ($key !== '')
{
$id = sprintf('%012d', $id);
//echo $key.$value."\n";
$db->put("$queuename-$id", $value);
$id++;
}
}
$db->out($queuename);
?>