Leadfyy.co
/
lanzu_api_2021-12-26
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

添加ssdb三方库，添加ssdb task线程应用。

master
Mike 5 years ago
parent
d9045f2a06
commit
ab0537a39a
6 changed files with 688 additions and 3 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 20
      app/Controller/TestController.php
  2. 622
      app/Libs/SimpleSSDB.php
  3. 39
      app/TaskWorker/SSDBTask.php
  4. 3
      composer.json
  5. 4
      config/autoload/server.php
  6. 3
      config/routes.php

20
app/Controller/TestController.php
View File

@ -6,6 +6,11 @@ namespace App\Controller;
use Hyperf\HttpServer\Contract\RequestInterface; use Hyperf\HttpServer\Contract\RequestInterface;
use Hyperf\HttpServer\Annotation\AutoController; use Hyperf\HttpServer\Annotation\AutoController;
use Hyperf\Utils\Coroutine;
use Hyperf\Utils\ApplicationContext;
use Hyperf\Task\TaskExecutor;
use Hyperf\Task\Task;
use App\TaskWorker\SSDBTask;
/** /**
* @AutoController() * @AutoController()
@ -15,9 +20,22 @@ use Hyperf\HttpServer\Annotation\AutoController;
class TestController extends AbstractController class TestController extends AbstractController
{ {
private $name = 'default action'; private $name = 'default action';
public function index1(RequestInterface $request) public function index1(RequestInterface $request)
{ {
$this->name = 'index1 action';
// $container = ApplicationContext::getContainer();
// $exec = $container->get(TaskExecutor::class);
// $result = $exec->execute(new Task([MethodTask::class, 'handle'], [Coroutine::id()]));
$client = ApplicationContext::getContainer()->get(SSDBTask::class);
$result = $client->exec("set","bar","1234");
$result = $client->exec("get","bar");
// $client = ApplicationContext::getContainer()->get(MethodTask::class);
// $result = $client->handle("set");
$this->name = 'index1 action '. $result;
return $this->name; return $this->name;
} }

622
app/Libs/SimpleSSDB.php
View File

@ -0,0 +1,622 @@
<?php
/**
* Copyright (c) 2012, ideawu
* All rights reserved.
* @author: ideawu
* @link: http://www.ideawu.com/
*
* SSDB PHP client SDK.
*/
namespace App\Libs;
use Exception;
class SSDBException extends Exception
{
}
class SSDBTimeoutException extends SSDBException
{
}
/**
* All methods(except *exists) returns false on error,
* so one should use Identical(if($ret === false)) to test the return value.
*/
class SimpleSSDB extends SSDB
{
function __construct($host, $port, $timeout_ms=2000){
parent::__construct($host, $port, $timeout_ms);
$this->easy();
}
}
class SSDB_Response
{
public $cmd;
public $code;
public $data = null;
public $message;
function __construct($code='ok', $data_or_message=null){
$this->code = $code;
if($code == 'ok'){
$this->data = $data_or_message;
}else{
$this->message = $data_or_message;
}
}
function __toString(){
if($this->code == 'ok'){
$s = $this->data === null? '' : json_encode($this->data);
}else{
$s = $this->message;
}
return sprintf('%-13s %12s %s', $this->cmd, $this->code, $s);
}
function ok(){
return $this->code == 'ok';
}
function not_found(){
return $this->code == 'not_found';
}
}
// Depricated, use SimpleSSDB instead!
class SSDB
{
private $debug = false;
public $sock = null;
private $_closed = false;
private $recv_buf = '';
private $_easy = false;
public $last_resp = null;
function __construct($host, $port, $timeout_ms=2000){
$timeout_f = (float)$timeout_ms/1000;
$this->sock = @stream_socket_client("[$host]:$port", $errno, $errstr, $timeout_f);
if(!$this->sock){
throw new SSDBException("$errno: $errstr");
}
$timeout_sec = intval($timeout_ms/1000);
$timeout_usec = ($timeout_ms - $timeout_sec * 1000) * 1000;
@stream_set_timeout($this->sock, $timeout_sec, $timeout_usec);
if(function_exists('stream_set_chunk_size')){
@stream_set_chunk_size($this->sock, 1024 * 1024);
}
}
function set_timeout($timeout_ms){
$timeout_sec = intval($timeout_ms/1000);
$timeout_usec = ($timeout_ms - $timeout_sec * 1000) * 1000;
@stream_set_timeout($this->sock, $timeout_sec, $timeout_usec);
}
/**
* After this method invoked with yesno=true, all requesting methods
* will not return a SSDB_Response object.
* And some certain methods like get/zget will return false
* when response is not ok(not_found, etc)
*/
function easy(){
$this->_easy = true;
}
function close(){
if(!$this->_closed){
@fclose($this->sock);
$this->_closed = true;
$this->sock = null;
}
}
function closed(){
return $this->_closed;
}
private $batch_mode = false;
private $batch_cmds = array();
function batch(){
$this->batch_mode = true;
$this->batch_cmds = array();
return $this;
}
function multi(){
return $this->batch();
}
function exec(){
$ret = array();
foreach($this->batch_cmds as $op){
list($cmd, $params) = $op;
$this->send_req($cmd, $params);
}
foreach($this->batch_cmds as $op){
list($cmd, $params) = $op;
$resp = $this->recv_resp($cmd, $params);
$resp = $this->check_easy_resp($cmd, $resp);
$ret[] = $resp;
}
$this->batch_mode = false;
$this->batch_cmds = array();
return $ret;
}
function request(){
$args = func_get_args();
$cmd = array_shift($args);
return $this->__call($cmd, $args);
}
private $async_auth_password = null;
function auth($password){
$this->async_auth_password = $password;
return null;
}
function __call($cmd, $params=array()){
$cmd = strtolower($cmd);
if($this->async_auth_password !== null){
$pass = $this->async_auth_password;
$this->async_auth_password = null;
$auth = $this->__call('auth', array($pass));
if($auth !== true){
throw new Exception("Authentication failed");
}
}
if($this->batch_mode){
$this->batch_cmds[] = array($cmd, $params);
return $this;
}
try{
if($this->send_req($cmd, $params) === false){
$resp = new SSDB_Response('error', 'send error');
}else{
$resp = $this->recv_resp($cmd, $params);
}
}catch(SSDBException $e){
if($this->_easy){
throw $e;
}else{
$resp = new SSDB_Response('error', $e->getMessage());
}
}
if($resp->code == 'noauth'){
$msg = $resp->message;
throw new Exception($msg);
}
$resp = $this->check_easy_resp($cmd, $resp);
return $resp;
}
private function check_easy_resp($cmd, $resp){
$this->last_resp = $resp;
if($this->_easy){
if($resp->not_found()){
return NULL;
}else if(!$resp->ok() && !is_array($resp->data)){
return false;
}else{
return $resp->data;
}
}else{
$resp->cmd = $cmd;
return $resp;
}
}
function multi_set($kvs=array()){
$args = array();
foreach($kvs as $k=>$v){
$args[] = $k;
$args[] = $v;
}
return $this->__call(__FUNCTION__, $args);
}
function multi_hset($name, $kvs=array()){
$args = array($name);
foreach($kvs as $k=>$v){
$args[] = $k;
$args[] = $v;
}
return $this->__call(__FUNCTION__, $args);
}
function multi_zset($name, $kvs=array()){
$args = array($name);
foreach($kvs as $k=>$v){
$args[] = $k;
$args[] = $v;
}
return $this->__call(__FUNCTION__, $args);
}
function incr($key, $val=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
function decr($key, $val=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
function zincr($name, $key, $score=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
function zdecr($name, $key, $score=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
function zadd($key, $score, $value){
$args = array($key, $value, $score);
return $this->__call('zset', $args);
}
function zRevRank($name, $key){
$args = func_get_args();
return $this->__call("zrrank", $args);
}
function zRevRange($name, $offset, $limit){
$args = func_get_args();
return $this->__call("zrrange", $args);
}
function hincr($name, $key, $val=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
function hdecr($name, $key, $val=1){
$args = func_get_args();
return $this->__call(__FUNCTION__, $args);
}
private function send_req($cmd, $params){
$req = array($cmd);
foreach($params as $p){
if(is_array($p)){
$req = array_merge($req, $p);
}else{
$req[] = $p;
}
}
return $this->send($req);
}
private function recv_resp($cmd, $params){
$resp = $this->recv();
if($resp === false){
return new SSDB_Response('error', 'Unknown error');
}else if(!$resp){
return new SSDB_Response('disconnected', 'Connection closed');
}
if($resp[0] == 'noauth'){
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
switch($cmd){
case 'dbsize':
case 'ping':
case 'qset':
case 'getbit':
case 'setbit':
case 'countbit':
case 'strlen':
case 'set':
case 'setx':
case 'setnx':
case 'zset':
case 'hset':
case 'qpush':
case 'qpush_front':
case 'qpush_back':
case 'qtrim_front':
case 'qtrim_back':
case 'del':
case 'zdel':
case 'hdel':
case 'hsize':
case 'zsize':
case 'qsize':
case 'hclear':
case 'zclear':
case 'qclear':
case 'multi_set':
case 'multi_del':
case 'multi_hset':
case 'multi_hdel':
case 'multi_zset':
case 'multi_zdel':
case 'incr':
case 'decr':
case 'zincr':
case 'zdecr':
case 'hincr':
case 'hdecr':
case 'zget':
case 'zrank':
case 'zrrank':
case 'zcount':
case 'zsum':
case 'zremrangebyrank':
case 'zremrangebyscore':
case 'ttl':
case 'expire':
if($resp[0] == 'ok'){
$val = isset($resp[1])? intval($resp[1]) : 0;
return new SSDB_Response($resp[0], $val);
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
case 'zavg':
if($resp[0] == 'ok'){
$val = isset($resp[1])? floatval($resp[1]) : (float)0;
return new SSDB_Response($resp[0], $val);
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
case 'get':
case 'substr':
case 'getset':
case 'hget':
case 'qget':
case 'qfront':
case 'qback':
if($resp[0] == 'ok'){
if(count($resp) == 2){
return new SSDB_Response('ok', $resp[1]);
}else{
return new SSDB_Response('server_error', 'Invalid response');
}
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
break;
case 'qpop':
case 'qpop_front':
case 'qpop_back':
if($resp[0] == 'ok'){
$size = 1;
if(isset($params[1])){
$size = intval($params[1]);
}
if($size <= 1){
if(count($resp) == 2){
return new SSDB_Response('ok', $resp[1]);
}else{
return new SSDB_Response('server_error', 'Invalid response');
}
}else{
$data = array_slice($resp, 1);
return new SSDB_Response('ok', $data);
}
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
break;
case 'keys':
case 'zkeys':
case 'hkeys':
case 'hlist':
case 'zlist':
case 'qslice':
if($resp[0] == 'ok'){
$data = array();
if($resp[0] == 'ok'){
$data = array_slice($resp, 1);
}
return new SSDB_Response($resp[0], $data);
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
case 'auth':
case 'exists':
case 'hexists':
case 'zexists':
if($resp[0] == 'ok'){
if(count($resp) == 2){
return new SSDB_Response('ok', (bool)$resp[1]);
}else{
return new SSDB_Response('server_error', 'Invalid response');
}
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
break;
case 'multi_exists':
case 'multi_hexists':
case 'multi_zexists':
if($resp[0] == 'ok'){
if(count($resp) % 2 == 1){
$data = array();
for($i=1; $i<count($resp); $i+=2){
$data[$resp[$i]] = (bool)$resp[$i + 1];
}
return new SSDB_Response('ok', $data);
}else{
return new SSDB_Response('server_error', 'Invalid response');
}
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
break;
case 'scan':
case 'rscan':
case 'zscan':
case 'zrscan':
case 'zrange':
case 'zrrange':
case 'hscan':
case 'hrscan':
case 'hgetall':
case 'multi_hsize':
case 'multi_zsize':
case 'multi_get':
case 'multi_hget':
case 'multi_zget':
case 'zpop_front':
case 'zpop_back':
if($resp[0] == 'ok'){
if(count($resp) % 2 == 1){
$data = array();
for($i=1; $i<count($resp); $i+=2){
if($cmd[0] == 'z'){
$data[$resp[$i]] = intval($resp[$i + 1]);
}else{
$data[$resp[$i]] = $resp[$i + 1];
}
}
return new SSDB_Response('ok', $data);
}else{
return new SSDB_Response('server_error', 'Invalid response');
}
}else{
$errmsg = isset($resp[1])? $resp[1] : '';
return new SSDB_Response($resp[0], $errmsg);
}
break;
default:
return new SSDB_Response($resp[0], array_slice($resp, 1));
}
return new SSDB_Response('error', 'Unknown command: $cmd');
}
function send($data){
$ps = array();
foreach($data as $p){
$ps[] = strlen($p);
$ps[] = $p;
}
$s = join("\n", $ps) . "\n\n";
if($this->debug){
echo '> ' . str_replace(array("\r", "\n"), array('\r', '\n'), $s) . "\n";
}
try{
while(true){
$ret = @fwrite($this->sock, $s);
if($ret === false || $ret === 0){
$this->close();
throw new SSDBException('Connection lost');
}
$s = substr($s, $ret);
if(strlen($s) == 0){
break;
}
@fflush($this->sock);
}
}catch(Exception $e){
$this->close();
throw new SSDBException($e->getMessage());
}
return $ret;
}
function recv(){
$this->step = self::STEP_SIZE;
while(true){
$ret = $this->parse();
if($ret === null){
try{
$data = @fread($this->sock, 1024 * 1024);
if($this->debug){
echo '< ' . str_replace(array("\r", "\n"), array('\r', '\n'), $data) . "\n";
}
}catch(Exception $e){
$data = '';
}
if($data === false || $data === ''){
if(feof($this->sock)){
$this->close();
throw new SSDBException('Connection lost');
}else{
throw new SSDBTimeoutException('Connection timeout');
}
}
$this->recv_buf .= $data;
# echo "read " . strlen($data) . " total: " . strlen($this->recv_buf) . "\n";
}else{
return $ret;
}
}
}
const STEP_SIZE = 0;
const STEP_DATA = 1;
public $resp = array();
public $step;
public $block_size;
private function parse(){
$spos = 0;
$epos = 0;
$buf_size = strlen($this->recv_buf);
// performance issue for large reponse
//$this->recv_buf = ltrim($this->recv_buf);
while(true){
$spos = $epos;
if($this->step === self::STEP_SIZE){
$epos = strpos($this->recv_buf, "\n", $spos);
if($epos === false){
break;
}
$epos += 1;
$line = substr($this->recv_buf, $spos, $epos - $spos);
$spos = $epos;
$line = trim($line);
if(strlen($line) == 0){ // head end
$this->recv_buf = substr($this->recv_buf, $spos);
$ret = $this->resp;
$this->resp = array();
return $ret;
}
$this->block_size = intval($line);
$this->step = self::STEP_DATA;
}
if($this->step === self::STEP_DATA){
$epos = $spos + $this->block_size;
if($epos <= $buf_size){
$n = strpos($this->recv_buf, "\n", $epos);
if($n !== false){
$data = substr($this->recv_buf, $spos, $epos - $spos);
$this->resp[] = $data;
$epos = $n + 1;
$this->step = self::STEP_SIZE;
continue;
}
}
break;
}
}
// packet not ready
if($spos > 0){
$this->recv_buf = substr($this->recv_buf, $spos);
}
return null;
}
}

39
app/TaskWorker/SSDBTask.php
View File

@ -0,0 +1,39 @@
<?php
declare(strict_types=1);
namespace App\TaskWorker;
use Hyperf\Utils\ApplicationContext;
use App\Libs\SimpleSSDB;
use Hyperf\Task\Annotation\Task;
class SSDBTask
{
/**
* @var SimpleSSDB
*/
public $ss = null;
/**
* @Task
*/
public function exec($method,...$args)
{
$result = $this->client()->__call($method,$args);
return $result;
}
protected function client(){
if ($this->ss instanceof SimpleSSDB) {
return $this->ss;
}
$this->ss = new SimpleSSDB(env('SSDB_HOST'), env('SSDB_PORT'));
$this->ss->auth(env('SSDB_AUTH'));
return $this->ss;
}
}

3
composer.json
View File

@ -29,7 +29,8 @@
"hyperf/model-cache": "~2.0.0", "hyperf/model-cache": "~2.0.0",
"hyperf/filesystem": "^2.0", "hyperf/filesystem": "^2.0",
"xxtime/flysystem-aliyun-oss": "^1.5", "xxtime/flysystem-aliyun-oss": "^1.5",
"hyperf/validation": "^2.0"
"hyperf/validation": "^2.0",
"hyperf/task": "^2.0"
}, },
"require-dev": { "require-dev": {
"swoole/ide-helper": "^4.5", "swoole/ide-helper": "^4.5",

4
config/autoload/server.php
View File

@ -36,10 +36,14 @@ return [
'max_request' => 100000, 'max_request' => 100000,
'socket_buffer_size' => 2 * 1024 * 1024, 'socket_buffer_size' => 2 * 1024 * 1024,
'buffer_output_size' => 2 * 1024 * 1024, 'buffer_output_size' => 2 * 1024 * 1024,
'task_worker_num' => 8,
'task_enable_coroutine' => false,
], ],
'callbacks' => [ 'callbacks' => [
SwooleEvent::ON_WORKER_START => [Hyperf\Framework\Bootstrap\WorkerStartCallback::class, 'onWorkerStart'], SwooleEvent::ON_WORKER_START => [Hyperf\Framework\Bootstrap\WorkerStartCallback::class, 'onWorkerStart'],
SwooleEvent::ON_PIPE_MESSAGE => [Hyperf\Framework\Bootstrap\PipeMessageCallback::class, 'onPipeMessage'], SwooleEvent::ON_PIPE_MESSAGE => [Hyperf\Framework\Bootstrap\PipeMessageCallback::class, 'onPipeMessage'],
SwooleEvent::ON_WORKER_EXIT => [Hyperf\Framework\Bootstrap\WorkerExitCallback::class, 'onWorkerExit'], SwooleEvent::ON_WORKER_EXIT => [Hyperf\Framework\Bootstrap\WorkerExitCallback::class, 'onWorkerExit'],
SwooleEvent::ON_TASK => [Hyperf\Framework\Bootstrap\TaskCallback::class, 'onTask'],
SwooleEvent::ON_FINISH => [Hyperf\Framework\Bootstrap\FinishCallback::class, 'onFinish'],
], ],
]; ];

3
config/routes.php
View File

@ -22,6 +22,7 @@ Router::addGroup('/v1/',function (){
Router::post('ServiceEvaluate/evaluate', 'App\Controller\ServiceEvaluateController@evaluate'); Router::post('ServiceEvaluate/evaluate', 'App\Controller\ServiceEvaluateController@evaluate');
Router::post('ServiceEvaluate/getPersonnelInfo', 'App\Controller\ServiceEvaluateController@getPersonnelInfo'); Router::post('ServiceEvaluate/getPersonnelInfo', 'App\Controller\ServiceEvaluateController@getPersonnelInfo');
Router::post('ServiceEvaluate/getEvaluateList', 'App\Controller\ServiceEvaluateController@getEvaluateList'); Router::post('ServiceEvaluate/getEvaluateList', 'App\Controller\ServiceEvaluateController@getEvaluateList');
Router::get('test/index1', 'App\Controller\TestController@index1');
}); });
Write Preview
Loading…
Cancel
Save