发表于

php-fpm

php-fpm master进程负责创建和管理woker进程，同时负责监听listen连接，master进程是多路复用的；woker进程负责accept请求连接，同时处理请求，一个woker进程可以处理多个请求（复用，不需要每次都fork一个woker进程），但一个woker进程一次只能处理一个请求（请区别理解和前一句话的意思）。以下伪代码说明一下关系：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
$socket = stream_socket_server('tcp://127.0.0.1:8080');
for ($i=0; $i < 5; $i++) { 
	if(pcntl_fork() == 0) {
		while (true) {
			// woker进程负责accept请求，处理请求
			$client = stream_socket_accept($socket);
			if(!$client) {
				continue;
			}
			$request = fread($client);
			$response = do_request($request);
			fwrite($client, $response);
			fclose($client);
		}
	}
}

好处是，复用worker进程，woker进程可以处理多个请求，试想一下，我们使用mysql、redis长连接的时候，连接状态是保存在woker进程的，这样后面的请求就可以直接使用，不需要每次都经历TCP握手。

验证

为了证实，一个woker进程一次只处理一个请求，可以把php-fpm配置的pm.max_children改成1，即只有一个woker进程处理请求，在这种请求下实现一个计数器，每个请求都加1，数据保存在文件中，读写文件不上文件锁，使用ab进行压测，查看最后结果加上失败次数是否和压测总请求量相等。

1
2
3
$counter = file_get_contents('./counter.txt');
$counter = $counter? ++$counter: 1;
file_put_contents('./counter.txt', $counter);

结语

php-fpm.conf有个配置项events.mechanism = epoll，指定事件驱动模型，但是这个是配置指定master进程的，不是woker进程的，以上例子可以看出，woker进程是单进程单线程模型。复用woker进程，处理多个请求好处是挺多的，每个woker进程启动和销毁，都需要执行一遍php扩展的模块初始化、销毁的工作，这部分开销还是挺大的。还有就是mysql、redis长连接等，因为php没有连接池，每次都执行短连接，TCP三次握手耗时，频繁关闭连接还会造成大量TIME_WAIT，造成服务器端口等资源大量消耗。

发表于

strstr

搜索字符串在另一字符串中的第一次出现，并返回剩余部分。

时间复杂度

O(n+m)

源码

strstr实现思路是，先查询子串的位置，返回该位置后的字符串。核心函数是 zend_memnstr_ex，使用的子串搜索算法是Sunday算法，这个具体就不说了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
// zend_memnstr_ex
zend_memnstr_ex_pre(td, needle, needle_len, 0); // 初始化跳转表
p = haystack;
end -= needle_len;
while (p <= end) {
	for (i = 0; i < needle_len; i++) {
		if (needle[i] != p[i]) {
			break;
		}
	}
	if (i == needle_len) {
		return p;
	}
	if (UNEXPECTED(p == end)) {
		return NULL;
	}
	p += td[(unsigned char)(p[needle_len])];  // 如果查询失败，通过跳转表决定搜索从搜索串什么位置开始
}
// zend_memnstr_ex_pre 构造跳转表
// Sunday算法重要的一部分，构造跳转表
// 跳转表是个256位长的数组
for (i = 0; i < 256; i++) {
	td[i] = needle_len + 1;
}
if (reverse) {
	for (i = needle_len - 1; i >= 0; i--) {
		td[(unsigned char)needle[i]] = i + 1;
	}
} else {
	for (i = 0; i < needle_len; i++) {
		// 例如子串是 aba
		// 最后跳转表，td = array('a' => 1, 'b' => 2, '其他' => 4)
		td[(unsigned char)needle[i]] = (int)needle_len - i;
	}
}

例子

1
strstr('bfabceabadde', 'aba');

1、zend_memnstr_ex_pre 构造跳转表，如下：

a b 其他
1 2 4

2、查询过程

b f a b c e a b a d d e
1 a b a
2 a b a
3 a b a 
1
2
3
4
5
第一步，b ！= a，源码第9行，跳出来；接着，源码19行，查找下一个查询从哪里开始；
p += td[(unsigned char)(p[needle_len])]，此时的p[needle_len]为'b'；
相当于，p += td['b']，看跳转表，p += 2，所以跳到了第二步的位置；
如果使用朴素算法，时间复杂度就是O(n*m)了。

发表于

setnx

Redis的setnx指令，设置一个键值对，当且仅当键不存在的时候，才能设置成功。

1
2
3
4
5
6
7
8
9
get test_setnx
$-1
setnx test_setnx 10
:1
get test_setnx
$2
10
setnx test_setnx 99
:0

用法

一般会使用setnx来实现锁的功能，解决资源竞争、缓存风暴等问题。例如，在缓存风暴中，没有锁保护的情况下，缓存失效，会导致短时间内，多个请求透过缓存到达数据库，请求同一份数据，修改同一份缓存；如果使用了锁，可以让获得锁的请求到达数据库，请求数据后回写缓存，后续没有得到锁的就直接读取新的缓存数据，而不用请求数据库了。

锁的实现

下面我一个一个坑地踩，一个一个坑地填，最终呈现完整的实现。

版本1

1
2
3
4
5
6
7
8
// 上锁
if($redis->setnx('lock_key', 1)){
	echo '成功'.PHP_EOL;
	// 释放锁
	$redis->delete('lock_key');
}else{
	echo '失败'.PHP_EOL;
}

初看貌似没有什么问题，但其实有很严重的问题，如果，在setnx成功后，请求挂掉了，或者忘了delete锁，那么’lock_key’这个锁就被永远锁着，出现死锁了，后续的就没法使用了，解决办法，增加个过期时间（版本2.1）？让它一段时间后自动销毁。

版本2.1

1
2
3
4
5
6
7
8
9
// 上锁
if($redis->setnx('lock_key', 1)){
	$redis->expire('lock_key', 1);
	echo '成功'.PHP_EOL;
	// 释放锁
	$redis->delete('lock_key');
}else{
	echo '失败'.PHP_EOL;
}

乍看没什么问题，其实问题还是存在，依旧是死锁的问题，只是问题的出现转移了，如果setnx成功，在expire前，请求挂掉，死锁出现。其实是，setnx和expire不是原子操作，那么用redis自带的事务操作会怎样呢？（版本2.2）

版本2.2

1
2
3
4
$redis->multi();
$redis->setnx('lock_key', 1);
$redis->expire('lock_key', 1);
$redis->exec();

用上事务，其实还是存在问题，那就是，请求过多的时候，锁的过期时间一直被更新，上锁那个家伙在手动释放锁之前退出了，就会导致锁一直有效。

其实以上几种情况讨论的都是没有正常释放锁的情况，保证不出现死锁，过期时间确是一个正确思路，至少官方给出的思路就是用过期时间辅助实现。只是实现的方式不一样（版本3）。

版本3

1
2
3
4
5
6
7
8
$now = get_millisecond();
$lock = $redis->setnx($lock_key,$lock_timeout);
if($lock or (($now > (float)$redis->get($lock_key)) and $now > (float)$redis->get_set($lock_key,$lock_timeout))) {
	echo '成功'.PHP_EOL;
}else{
	echo '失败'.PHP_EOL;
}

上版本使用setnx和get_set来实现。做法和上三个版本有什么不一样呢，setnx保存的value不是一个true or 1，而是过期的绝对时间戳，为什么这么做呢。

来，我们回到死锁的情况，锁没有被释放掉，以后的请求setnx都会失败，这时候会进入第二步判断，判断锁是否超时失效了($now > get($lock_key))，这时候get_set要出场了，看下下面的场景：

1
2
3
4
5
6
A setnx成功，过期时间戳为5（这是绝对时间戳，为了简化阅读）；
A没有delete锁，挂了；
当前时间戳为6($now > get($lock_key)；6 > 5)，那就是A设置的锁过期了；
B请求锁，过期时间戳为10；同时C也请求锁，过期时间戳为9；
假如C请求成功，C get_set 设置成9，返回5，$now == 6,6 > 5,所以C获得锁；
在C请求锁后，B也请求锁（慢一步，命运的安排），B get_set 设置成10，返回9，$now == 6,6 < 9,表示锁有效，已经被别人获得并更新了，B没有获得锁；

这里在别人获得锁后，也被更新过期时间，但不像版本2.1那样，谁都会频繁更新过期时间，所以不会出现版本2.1那样锁长期有效。

结语

这样的锁，肯定不是什么高大上的方法，但成本低，效果不错，满足大部分需求了，在我司的抽奖系统里很多地方都可以看到，经过线上考验。如果想要高大上的实现，可以考虑google chubby。

参考：Redis setnx
github：github源码

发表于

in_array

函数搜索数组中是否存在指定的值。存在，返回true；否，返回false。

时间复杂度

最差是O(n)，平均O(n/2)

源码

只拿一个分支出来看，其他的差不多，从源码中的ZEND_HASH_FOREACH_KEY_VAL，可以看出实现方式，是遍历数组，搜索是否存在这个值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
ZEND_HASH_FOREACH_KEY_VAL(Z_ARRVAL_P(array), num_idx, str_idx, entry) {
	if (fast_equal_check_long(value, entry)) {
		if (behavior == 0) {
			RETURN_TRUE;
		} else {
			if (str_idx) {
				RETVAL_STR_COPY(str_idx);
			} else {
				RETVAL_LONG(num_idx);
			}
			return;
		}
	}
} ZEND_HASH_FOREACH_END();

优化

网上有人给出两种优化方法。第一种，使用 array_flip key-value翻转，通过isset（或者array_key_exists）判断是否存在；第二种，通过implode把数组合并成一个字符串，再通过strpos搜索。对于这两种优化，我是持有怀疑态度的。下面一一来试验一番。

先看in_array耗时，经过多次测试，耗时100w整型在30ms左右：

1
2
3
4
5
6
7
8
9
10
ini_set('memory_limit', '2048M');
$arr = range(0,10000000,1);
echo get_millisecond().' ms'.PHP_EOL;
var_dump(in_array(10000000, $arr));
echo get_millisecond().' ms'.PHP_EOL;
// 1499311572333 ms
// bool(true)
// 1499311572363 ms

第一种，先翻转，再isset；经过多次测试，耗时100w整型在200ms左右：

1
2
3
4
5
6
7
8
9
10
11
ini_set('memory_limit', '2048M');
$arr = range(0,10000000,1);
echo get_millisecond().' ms'.PHP_EOL;
$new_arr = array_flip($arr);
var_dump(isset($new_arr[10000000]));
echo get_millisecond().' ms'.PHP_EOL;
// 1499311825323 ms
// bool(true)
// 1499311825563 ms

第二种，先implode成一个字符串，再strpos搜索；经过多次测试，耗时100w整型在400ms左右：

1
2
3
4
5
6
7
8
9
10
11
12
ini_set('memory_limit', '2048M');
$arr = range(0,10000000,1);
echo get_millisecond().' ms'.PHP_EOL;
$s = implode('_', $arr);
$s = '_'.$s.'_';
var_dump(strpos($s, '_10000000_'));
echo get_millisecond().' ms'.PHP_EOL;
// 1499312200167 ms
// int(78888890)
// 1499312200624 ms

优化总结

第一种，iseet，hash表的查找，在良好的hash算法下，时间复杂度可以达到O(1)，但不要忘记了，在isset前还有个，array_flip操作，这个的时间复杂度是也O(n)，还需要付出额外的内存空间，毕竟生成了一个新的数组。

第二种，strpos php底层的实现是KMP算法，时间复杂度是O(n+m)，而且还有implode的操作，最后还会生成一个字符串，也需要付出额外的内存空间。

那么，是否in_array就已经没法优化了呢，其实也不然，这个的优化不能撇开使用场景。如果是需要频繁搜索的，可以一开始就构建一个集合，例如：array(search => true)，这样可以直接用isset；如果是需要一个有序数组，那么一开始就构建成有序数组，array(0,1,2,3)，搜索可以用上搜索算法，快排等。