多进程加锁入库实例

追影

1. <?php
   
2. 
   
3. class caijitodb extends Controller
   
4. {
   
5.     public function run($params = null)
   
6.     {
   
7.         ini_set('memory_limit', '2048M');
   
8.         $this->config->set('config_weight_class_id', 1);
   
9.         $this->config->set('config_language', 'en-gb');
   
10.         $this->config->set('config_language_id', 1);
    
11.         $this->config->set('config_length_class_id', 1);
    
12.         try {
    
13.             $start_id=0;
    
14.             $start_file = __DIR__ . DIRECTORY_SEPARATOR . 'todb.txt';
    
15.             if (is_file($start_file)) {
    
16.                 clearstatcache();
    
17.                 a:
    
18.                 $fp = fopen($start_file, 'r+');
    
19.                 $start_id = @fgets($fp);
    
20.                 var_dump($start_id);
    
21.                 if ($start_id === false) {
    
22.                     sleep(1);
    
23.                     echo '等待解锁'.PHP_EOL;
    
24.                     goto a;
    
25.                 }
    
26.                 flock($fp, LOCK_EX);// 加锁
    
27.                 //$start_id= (int)file_get_contents($start_file);
    
28.             } else {
    
29.                 echo $start_file.'文件不存在';
    
30.                 exit();
    
31.             }
    
32.             $start_id = intval($start_id);
    
33.             $sql = "select * from i8_pipe where status=0 and id>{$start_id} order by id asc limit 100";
    
34.             $query = $this->db->query($sql);
    
35.             $rows = $query->rows;
    
36.             if ($rows) {
    
37.                 $ids = [];
    
38.                 $total_num = count($rows);
    
39.                 $last_num = ($total_num-1)*1;
    
40.                 $start_id = $rows[$last_num]['id'];
    
41.                 fseek($fp, 0);
    
42.                 fwrite($fp, $start_id);
    
43.                 flock($fp, LOCK_UN);// 解锁
    
44.                 fclose($fp);
    
45.                 foreach ($rows as $k => $row) {
    
46.                     $data = json_decode($row['data'], true);
    
47.                     if ($data) {
    
48.                         try {
    
49.                             $r = $this->load->controller('api/product/add', $data);
    
50.                             if ($r) {
    
51.                                 $ids[] = $row['id'];
    
52.                                 echo time() . " " . $row['id'] . " into db success" . PHP_EOL;
    
53.                             } else {
    
54.                                 echo time() . " " . $row['id'] . " fail" . PHP_EOL;
    
55.                             }
    
56.                         }catch (\Exception $exception){
    
57.                             echo time() . " " . $row['id'] . " fail" . PHP_EOL;
    
58.                         }
    
59.                     }else{
    
60.                         $sql = "delete from i8_pipe where id = ".$row['id'];
    
61.                         $this->db->query($sql);
    
62.                     }
    
63.                 }
    
64.                 if ($ids) {
    
65.                     $sql = "update i8_pipe set status=1 where id in(" . join(',', $ids) . ")";
    
66.                     $this->db->query($sql);
    
67.                     echo time().' success'.PHP_EOL;
    
68.                 }
    
69.             } else {
    
70.                 flock($fp, LOCK_UN);// 解锁
    
71.                 fclose($fp);
    
72.                 echo 'finished';
    
73.                 exit;
    
74.             }
    
75.             $this->run($params);
    
76.         } catch (\Exception $exception) {
    
77.             echo $exception->getMessage().PHP_EOL;
    
78.             if($fp){
    
79.                 flock($fp, LOCK_UN);// 解锁
    
80.                 fclose($fp);
    
81.             }
    
82.         }
    
83.     }
    
84. }
    

复制代码