php多线程实现多进程 跨平台的一例代码

3. define('DIR_PHP_EXEC', 'php');

4. define('DIR_MAIN_EXEC', __FILE__);
5. define('DIR_TMP', '/tmp');
6. require_once('my_process.php');

7. class pp extends my_process_base {

8. public function run($param = null) {
9. for ($i = 0; $i echo "111 $param\n";
10. sleep(1);
11. }
12. }
13. }

14. init_my_process();

15. $obj = $GLOBALS['gal_obj_process_m'];
16. if ($obj->is_main()) {
17. $obj->run_task('pp', 'a');
18. $obj->run_task('pp', 'b');
19. $obj->run_task('pp', 'c');
20. $obj->run_task('pp', 'd');
21. //$obj->run_task('pp', 'b');
22. $obj->set_max_run(10);
23. $obj->run();
24. }?>

复制代码

2，进程管理类

3. /**

4. * @copyright 2007 movivi
5. * @author 徐智 *
6. * $Id: getPage.php 11 2007-09-21 02:15:01Z fred $
7. * modify by bbs.it-home.org
8. */
9. if (!defined('DIR_PHP_EXEC')) define('DIR_PHP_EXEC', 'php');
10. //if (!defined('DIR_MAIN_EXEC')) define('DIR_MAIN_EXEC', '');
11. if (!defined('DIR_TMP')) define('DIR_TMP', '');
12. /********/
13. /* 初始化 */
14. define('CMD_MAIN_PROCESS_KEY', 'main_process_key');
15. define('CMD_CHILD_PROCESS_NAME', 'child_process_name');
16. define('CMD_CHILD_PROCESS_PARAM', 'child_process_param');

17. function init_my_process() {

18. $GLOBALS['gal_obj_cmd'] = new my_cmd_argv();
19. $key = $GLOBALS['gal_obj_cmd']->get_value(CMD_MAIN_PROCESS_KEY);
20. $key = $key === false ? '' : $key;
21. $GLOBALS['gal_obj_process_m'] = new my_process_m($key);
22. if (!$GLOBALS['gal_obj_process_m']->is_main()) $GLOBALS['gal_obj_process_m']->run() ;
23. }

24. /**

25. * php多进程类
26. *
27. * 你需要从这个对象继承，然后实现你自己的run处理
28. */
29. abstract class my_process_base {
30. public function __construct($auto_run=true, $name='') {
31. }

32. public function __destruct() {

33. echo "@end\n";
34. }

35. abstract public function run($param = null);

36. }
38. class my_cmd_argv {
39. private $cmd_argv = array();
40. public function __construct() {
41. $argv = $_SERVER['argv'];
42. for ($i = 1; $i $cmd = explode('=', $argv[$i]);
43. $this->cmd_argv[$cmd[0]] = isset($cmd[1]) ? $cmd[1] : '';
44. }
45. }

46. public function get_key($key) {

47. return isset($this->cmd_argv[$key]);
48. }

49. public function get_value($key) {

50. return isset($this->cmd_argv[$key]) ? $this->cmd_argv[$key] : false;
51. }
52. }

53. /**

54. * php多进程管理类
55. * 可以在PHP中实现多进程处理，只限在控制台方式使用
56. * 当前的信号实现机制采用文件方式
57. *
58. */
59. class my_process_m {
60. /**
61. * @var array $task_list
62. * 进程列表
63. */
64. private $task_list = array();
65. private $lock_list = array();
66. private $lock = null;
67. private $is_main = false;
68. private $max_run = 3600000;

69. private function release_lock($key = null) {

70. $lock = &$this->lock_list;
71. if (!is_null($key)) {
72. $key = md5($this->build_lock_id($key));
73. if (isset($lock[$key])) {
74. if (is_resource($lock[$key][0])) fclose($lock[$key][0]);
75. unlink($lock[$key][1]);
76. unset($lock[$key]);
77. }
78. return true;
79. }

80. foreach ($lock as $k => $h) {

81. if (is_resource($h)) fclose($h);
82. unset($lock[$k]);
83. }
84. return true;
85. }

86. private function release_task($key = null) {

87. $task = &$this->task_list;
88. if (!is_null($key) && isset($task[$key])) {
89. if (is_resource($task[$key])) pclose($task[$key]);
90. unset($task[$key]);
91. } else {
92. foreach ($task as $k => $h) {
93. if (is_resource($h)) pclose($h);
94. unset($task[$k]);
95. }
96. }
97. return true;
98. }
100. private function build_lock_id($key) {
101. return DIR_TMP . DIRECTORY_SEPARATOR . $key . '_sem.lock';
102. }

103. protected function run_child_process() {

104. $class = $GLOBALS['gal_obj_cmd']->get_value(CMD_CHILD_PROCESS_NAME);
105. $param = $GLOBALS['gal_obj_cmd']->get_value(CMD_CHILD_PROCESS_PARAM);
106. $param = $param == '' ? null : unserialize(base64_decode(trim($param)));
107. $obj = new $class();
108. $obj->run($param);
109. $this->task_list[] = $obj;
110. }

111. public function __construct($lock='') {

112. if ($lock === '') {
113. $this->is_main = true;
114. $key = md5(uniqid()) . '_main.my_process';
115. $lock = array($key, $this->get($key));
116. } else {
117. $this->is_main = false;
118. $lock = array($lock, 0);
119. }
120. $this->lock = $lock;
121. }

122. public function __destruct() {

123. $this->release_lock();
124. $this->release_task();
125. }

126. /**

127. * 停止所有进程
128. *
129. */
130. public function stop_all() {
131. }

132. /**

133. * 是否是主进程
134. *
135. */
136. public function is_main() {
137. return $this->is_main;
138. }

139. /**

140. * 是不是已经存在一个活动信号
141. *
142. * @param string $key
143. * @return bool
144. */
145. public function exist($key) {
146. return file_exists($this->build_lock_id($key));
147. }

148. /**

149. * 获取一个信号
150. *
151. * @param string $key
152. * @param int $max_acquire 最大请求阻塞数量
153. * @return mix 如果成功返回一个信号ID
154. *
155. */
156. public function get($key, $max_acquire=5) {
157. $fn = $this->build_lock_id($key);
158. if (isset($this->lock_list[md5($fn)])) return false;
159. $id = fopen($fn, 'a+');
160. if ($id) $this->lock_list[md5($fn)] = array($id, $fn);
161. return $id;
162. }

163. /**

164. * 释放一个信号
165. *
166. * @param string $key
167. * @return bool 如果成功返回一个信号true
168. *
169. */
170. public function remove($key) {
171. return $this->release_lock($key);
172. }

173. /**

174. * 获取一个信号
175. *
176. * @param string $id 信号ID
177. * @param bool $block 是否阻塞
178. */
179. public function acquire($id, $block=false) {
180. if ($block) {
181. return flock($id, LOCK_EX);
182. } else {
183. return flock($id, LOCK_EX + LOCK_NB);
184. }
185. }

186. /**

187. * 释放一个信号
188. *
189. */
190. public function release($id) {
191. flock($id, LOCK_UN);
192. }
193. public function run_task($process_name, $param=null) {
194. $this->task_list[] = popen(DIR_PHP_EXEC . ' -f ' . DIR_MAIN_EXEC . ' -- '
195. . CMD_CHILD_PROCESS_NAME . '=' . $process_name . ' '
196. . CMD_CHILD_PROCESS_PARAM . '="' . base64_encode(serialize($param)) . '" '
197. . CMD_MAIN_PROCESS_KEY . '="' . $this->lock[0] . '" ',
198. 'r');
199. }

200. public function run($auto_run = true) {

201. if ($this->is_main) {
202. $ps = &$this->task_list;
203. $max_run = &$this->max_run;
204. $id = 0;
205. do {
206. //echo "process-------------: \n";
207. $c = 0;
208. foreach ($ps as $k => $h) {
209. $c++;
210. $msg = fread($h, 8000);
211. if (substr($msg, -5, 4) === '@end') {
212. echo "end process:[$k][$id] echo \n{$msg} \n";
213. $this->release_task($k);
214. } else {
215. echo "process:[$k][$id] echo \n{$msg} \n";
216. }
217. }
218. sleep(1);
219. } while ($auto_run && $id++ 0);
220. } else {
221. $this->run_child_process();
222. }
223. }

224. public function set_max_run($max=1000) {

225. $this->max_run = $max;
226. }
227. }
228. ?>

复制代码