<?php
// 生产者 不断的生产大量数据 但是总会有停止的时候(本业务功能结束)
for ($i=100000; $i < 500000; $i++) {
  $file = "./nodes/{$i}.txt";
  touch($file);
  xattr_set($file, ‘dataCount‘, $i);
}

<?php
// 消费者 不断的读取生产的数据进行处理或者数据落地 没有停止的时候(为了避免与生产者有并发读写 自己会按时休息一会儿)

is_dir("./nodes") || mkdir("./nodes");

while (1) {
  $result = [];
  $list = glob("./nodes/*.txt");
  if ($list) {
    $list = array_slice($list, 0, 10000);
    foreach ($list as $file) {
      $result[] = xattr_get($file, ‘dataCount‘);
      unlink($file);
    }
    file_put_contents(‘data.log‘,"\r\n" . implode("\r\n", $result), FILE_APPEND);
  }
  usleep(1000);//休息1ms
}