$this->logger = $logger;

}

protected $logger;

protected static $dbh;
public function __construct() {

}
public function run(){
$dbhost = '192.168.2.1'; // 数据库服务器
$dbport = 3306;
$dbuser = 'www'; // 数据库用户名
$dbpass = 'qwer123'; // 数据库密码
$dbname = 'example'; // 数据库名

self::$dbh = new PDO("mysql:host=$dbhost;port=$dbport;dbname=$dbname",$dbuser,$dbpass,array(
/ PDO::MYSQL_ATTR_INIT_COMMAND => 'SET NAMES 'UTF8'',/
PDO::MYSQL_ATTR_COMPRESS => true,PDO::ATTR_PERSISTENT => true
)
);

}
protected function getInstance(){
return self::$dbh;
}

}

/ the collectable class implements machinery for Pool::collect /
class Fee extends Stackable {
public function __construct($msg) {
$trades = explode(",",$msg);
$this->data = $trades;
print_r($trades);
}

public function run() {

$this->worker->logger->log("%s executing in Thread #%lu",CLASS,$this->worker->getThreadId() );

try {
$dbh = $this->worker->getInstance();

$insert = "INSERT INTO fee(ticket,login,volume,status) VALUES(:ticket,:login,:volume,'N')";
$sth = $dbh->prepare($insert);
$sth->bindValue(':ticket',$this->data[0]);
$sth->bindValue(':login',$this->data[1]);
$sth->bindValue(':volume',$this->data[2]);
$sth->execute();
$sth = null;

/ ...... /

$update = "UPDATE fee SET status = 'Y' WHERE ticket = :ticket and status = 'N'";
$sth = $dbh->prepare($update);
$sth->bindValue(':ticket',$this->data[0]);
$sth->execute();
//echo $sth->queryString;
//$dbh = null;
}
catch(PDOException $e) {
$error = sprintf("%s,%sn",$mobile,$id );
file_put_contents("mobile_error.log",$error,FILE_APPEND);
}
}
}

class Example {
/ config /
const LISTEN = "tcp://192.168.2.15:5555";
const MAXCONN = 100;
const pidfile = CLASS;
const uid = 80;
const gid = 80;

protected $pool = NULL;
protected $zmq = NULL;
public function __construct() {
$this->pidfile = '/var/run/'.self::pidfile.'.pid';
}
private function daemon(){
if (file_exists($this->pidfile)) {
echo "The file $this->pidfile exists.n";
exit();
}

$pid = pcntl_fork();
if ($pid == -1) {
die('could not fork');
} else if ($pid) {
// we are the parent
//pcntl_wait($status); //Protect against Zombie children
exit($pid);
} else {
// we are the child
file_put_contents($this->pidfile,getmypid());
posix_setuid(self::uid);
posix_setgid(self::gid);
return(getmypid());
}
}
private function start(){
$pid = $this->daemon();
$this->pool = new Pool(self::MAXCONN,ExampleWorker::class,[]);
$this->zmq = new ZMQSocket(new ZMQContext(),ZMQ::SOCKET_REP);
$this->zmq->bind(self::LISTEN);

/ Loop receiving and echoing back /
while ($message = $this->zmq->recv()) {
//print_r($message);
//if($trades){
$this->pool->submit(new Fee($message));
$this->zmq->send('TRUE');
//}else{
// $this->zmq->send('FALSE');
//}
}
$pool->shutdown();
}
private function stop(){

if (file_exists($this->pidfile)) {
$pid = file_get_contents($this->pidfile);
posix_kill($pid,9);
unlink($this->pidfile);
}
}
private function help($proc){
printf("%s start | stop | help n",$proc);
}
public function main($argv){
if(count($argv) < 2){
printf("please input help parametern");
exit();
}
if($argv[1] === 'stop'){
$this->stop();
}else if($argv[1] === 'start'){
$this->start();
}else{
$this->help($argv[0]);
}
}
}

$cgse = new Example();
$cgse->main($argv);

$pid = pcntl_fork();
if ($pid == -1) {
die('could not fork');
} else if ($pid) {
// we are the parent
//pcntl_wait($status); //Protect against Zombie children
exit($pid);
} else {
// we are the child
file_put_contents($this->pidfile,getmypid());
posix_setuid(self::uid);
posix_setgid(self::gid);
return(getmypid());
}
}