pthreads v3下的同步处理synchronized

时间:2023-05-09 11:51:26

之所以会用到同步,是因为如果多个线程中对同一个资源进行操作时,会发生混乱。

比如2个线程对变量进行加1操作,第1个线程还没来的及改写数据时,第2个线程就对变量进行操作了,那变量最终的结果就是未知的,这个时候就需要同步来进行控制了。

例子如下:

<?php

class Count extends Thread
{
public $cnt = 0; public function run()
{
$this->add();
} public function add()
{
//对成员进行加1操作
for ($i = 0; $i < 100000; $i++) {
++$this->cnt;
}
}
} $c = new Count();
//调用start()时,线程run()中就调用了add()方法
$c->start();
//我们人为再调用一次add()方法,这时候就会有两个for循环对$cnt进行操作
$c->add();
//把创建的线程加入主线程中,让主线程等待子线程运行结束
$c->join(); //这里输出就是不确定性的
var_dump($c->cnt);

多次运行后,$cnt的值是不确定的。如下图所示:

pthreads v3下的同步处理synchronized

在pthreads v2中我们可以用Mutex,不过在v3版本中被删除了,所以我们可以简单的把加1操作放到synchronized中进行同步,代码如下:

<?php

class Count extends Thread
{
public $cnt = 0; public function run()
{
$this->add();
} public function add()
{
$this->synchronized(function () {
//对成员进行加1操作
for ($i = 0; $i < 100000; $i++) {
++$this->cnt;
}
});
}
} $c = new Count();
//调用start()时,线程run()中就调用了add()方法
$c->start();
//我们人为再调用一次add()方法,这时候就会有两个for循环对$cnt进行操作
$c->add();
//把创建的线程加入主线程中,让主线程等待子线程运行结束
$c->join(); //这里就会一直输出200000
var_dump($c->cnt);

结果如下所示:

pthreads v3下的同步处理synchronized

当然我们也可以通过notify()和wait()进行同步控制,代码如下:

<?php

class Task extends Thread
{
public $flag = 1; public function run()
{
$this->synchronized(function () {
//标识不为1就一直等待
if ($this->flag !== 1) {
$this->wait();
} for ($i = 1; $i <= 10; $i++) { echo "flag : {$this->flag} i : {$i} \n"; if ($this->flag === 1) {
//设置标识
$this->flag = 2;
//发送唤醒通知,然后让当前线程等待
//注意,notify()与wait()顺序不要搞错了,不然会一直阻塞
$this->notify();
$this->wait();
}
} //我们在这里再次调用notify()
//因为在最后一次输出flag : 2 i : 20时,当前线程的i已经变成11了,跳出了for循环,
//但另一个线程则一直阻塞在wait()那里,程序无法结束,所以需要notify()再次唤醒一次
$this->notify();
});
}
} $t = new Task();
$t->start(); $t->synchronized(function ($obj) {
//标识不为2就一直等待
if ($obj->flag !== 2) {
$obj->wait();
} for ($i = 11; $i <= 20; $i++) { echo "flag : {$obj->flag} i : {$i} \n"; if ($obj->flag === 2) {
$obj->flag = 1;
$obj->notify();
$obj->wait();
}
}
}, $t); //把创建的线程加入主线程中,让主线程等待子线程运行结束
$t->join();

结果如下图所示:

pthreads v3下的同步处理synchronized

我们通过notify()和wait()控制了两个for循环,来回的输出变量i的值,保证了顺序性。

我们再来看一个复杂点的例子,共享的资源,如果不进行同步操作,会出现不可预知的情况,代码如下:

<?php

class Task extends Thread
{
private $name;
private $file; public function __construct($name, $file)
{
$this->name = $name;
$this->file = $file;
} public function run()
{
$data = file_get_contents($this->file);
$data = floatval($data);
for ($i = 0; $i < 100000; $i++) {
++$data;
}
file_put_contents($this->file, $data);
echo "task : {$this->name} data : {$data} \n";
}
} $tasks = [];
$file = './test.log'; for ($i = 0; $i < 100; $i++) {
$tasks[$i] = new Task($i, $file);
$tasks[$i]->start();
} for ($i = 0; $i < 100; $i++) {
$tasks[$i]->join();
}

我们开100个线程对文件test.log进行读写,理想状态下,test.log中的数据应该是每次增加10000000的。现在的电脑配置都比较好,大家可以多运行几次就可以看出效果。

pthreads v3下的同步处理synchronized

很明显最后的数据好像少了200000,多线程下对test.log文件进行读写,而我们又没有加锁,显然是会出现数据混乱的。

现在我们修改一下代码,如下:

<?php

class File extends Thread
{
private $file; public function __construct($file)
{
$this->file = $file;
} public function inc()
{
//进行同步控制,当100个task线程调用inc方法时,synchronized可以保证块内的代码是同步的
//注意,注意,不要把inc方法写到Task里,那样是没效果的,因为每个task线程都是独立空间,他们各自调各自的inc方法,是没法做到同步的
//常用的做法是我们要同步哪些资源,就为这些资源写个Thread类,并提供操作这些资源的方法,并在方法里加上synchronized
return $this->synchronized(function () {
$data = file_get_contents($this->file);
$data = floatval($data);
for ($i = 0; $i < 100000; $i++) {
++$data;
}
file_put_contents($this->file, $data);
return $data;
});
}
} class Task extends Thread
{
private $name;
private $file; public function __construct($name, $file)
{
$this->name = $name;
$this->file = $file;
} public function run()
{
$data = $this->file->inc();
echo "task : {$this->name} data : {$data} \n";
}
} $tasks = [];
$file = new File('./test.log'); for ($i = 0; $i < 100; $i++) {
$tasks[$i] = new Task($i, $file);
$tasks[$i]->start();
} for ($i = 0; $i < 100; $i++) {
$tasks[$i]->join();
}

结果如下图所示,当然为了保险起见,我们可以试着多运行几次,下面是我运行了25次的结果:

pthreads v3下的同步处理synchronized