简单的 schdule Nodejs 脚本在连接到远程数据库时插入记录两次

我有一个时间表nodejs脚本。基本上，它有时间表和命令表。调度表有很多命令行。 Nodejs 脚本每 5 秒检查一次调度表。如果预定时间与当前时间匹配。我将命令行（从计划表）插入到命令表中。

错误是

我在笔记本电脑上运行脚本并进行测试数据库的本地副本在我的笔记本电脑上。 Asingle仅插入计划表中的命令once.

我在笔记本电脑上运行脚本并进行测试RDS（远程mysql服务器）在我的笔记本电脑上。 Asingle仅插入计划表中的命令twice，但其中之一，SchduleId 是null. (为什么 ScheduleId 为 NULL？)

完整代码

var config = require("./config.js");
var Promise = require('bluebird');
var mysql = require('promise-mysql');
var ON_DEATH = require('death');


var g_pool = null;

function connect_db() {
  g_pool = mysql.createPool(config.db_config);
}


function close_db() {
  g_pool.end(function (err) {
    // all connections in the pool have ended
    });
}



// http://thecodeship.com/web-development/alternative-to-javascript-evil-setinterval/
function interval(func, wait, times){
    var interv = function(w, t) {
    return function() {
        if(typeof t === "undefined" || t-- > 0) {
        setTimeout(interv, w);
        try {
            func.call(null);
        }
        catch(e) {
            t = 0;
          throw e.toString();
        }
      }
        };
    }(wait, times);

  setTimeout(interv, wait);
}


function get_current_utc_time() {
  var curr_date_obj = new Date();
    var time_utc = "";

    // somehow the date format is not accurate.
  //var time_utc = dateFormat(now, "yyyy-mm-dd h:MM:ss", true);

    var year = curr_date_obj.getUTCFullYear(); 
    var month = add_zero(curr_date_obj.getUTCMonth() + 1); // count from 0
    var date = add_zero(curr_date_obj.getUTCDate()); // count from 1
    var hr = add_zero(curr_date_obj.getUTCHours());
    var min = add_zero(curr_date_obj.getUTCMinutes()); 
    // we ignore the second
    var sec = "00";

    time_utc = year + "-" + month + "-" + date + " " + hr + ":" + min + ":" + sec;

    console.log("-current utc-");
    console.log(time_utc);

  return time_utc;
};


// http://www.w3schools.com/jsref/jsref_getutchours.asp
function add_zero(i) {
  if (i < 10) {
    i = "0" + i;
  }
  return i;
}


function insert_into_pending_cmd(msg_obj) {
  console.log();
  console.log("-insert_into_pending_cmd-");

    var schedule_id = msg_obj.schedule_id;
    var device_name = msg_obj.device_name;
    var cmd = msg_obj.cmd;
    var is_pending = msg_obj.is_pending;

  if(is_pending) {
    return Promise.resolve();
  }
  else {
    var curr_time = get_current_utc_time();
    var sql = "insert into Command set CommandDate = " + "'" + curr_time + "'" + "," + "RemoteName = " + "'" + device_name + "'" + "," + "CommandJSON = " + "'" + cmd + "'" + "," + "CommandComplete = 0" + "," + "ScheduleId = " + "'" + schedule_id + "'";

    return g_pool.query(sql).then(function(){
      return Promise.resolve();
    });
  }
}


function is_schedule_cmd_already_pending(msg_obj) {
    console.log();
  console.log("-is_schedule_cmd_already_pending-");

    var schedule_id = msg_obj.schedule_id;
    var device_name = msg_obj.device_name;
    var cmd = msg_obj.cmd;
    var is_run = msg_obj.is_run;

    var local_msg_obj = {};

  if(is_run) {
    var sql = "select count(*) as num from Command where ScheduleId = " + "'" + schedule_id + "'" + " and CommandComplete = 0 and (UNIX_TIMESTAMP(UTC_TIMESTAMP()) - UNIX_TIMESTAMP(CommandDate)) < 600 and (UNIX_TIMESTAMP(UTC_TIMESTAMP()) - UNIX_TIMESTAMP(CommandDate)) > 0";
    return g_pool.query(sql).then(function(rows){
      var num = rows[0].num;
      if(num == 0) {
                local_msg_obj = {
                    schedule_id: schedule_id,
                    device_name: device_name,
                    cmd: cmd,
                    is_pending: false
                };
        return Promise.resolve(local_msg_obj);
      }
      else {
                local_msg_obj = {
          schedule_id: schedule_id,
          device_name: device_name,
          cmd: cmd,
          is_pending: true 
        };
        return Promise.resolve(local_msg_obj);
      }
    });
  }
  else {
        local_msg_obj = {
        schedule_id: schedule_id,
      device_name: device_name,
      cmd: cmd,
      is_pending: true 
    };
    return Promise.resolve(local_msg_obj);
  }
}




function is_matchup_schedule_time(row) {
    // get all field
  var schedule_id = row.ScheduleId;
  var device_name = row.ScheduleRemoteName;
  var day_code = row.ScheduleDaycode;
  var schedule_time = row.ScheduleTime;
  var cmd = row.ScheduleCommandJSON;

  // get hour and min
  var schedule_time_arr = schedule_time.split(":");
  var schedule_hour = schedule_time_arr[0];
  var schedule_min = schedule_time_arr[1];

  // print
  console.log();
  console.log();
  console.log("- schedule_id, device_name, day_code, schedule_time, schedule_hr, schedule_min, cmd -");
  console.log(schedule_id);
  console.log(device_name);
  console.log(day_code);
  console.log(schedule_time);
  console.log(schedule_hour);
  console.log(schedule_min);
  console.log(cmd);

  // curr date obj
  var curr_date_obj = new Date();
  var curr_date_code = add_zero(curr_date_obj.getUTCDay());

    // print current
  console.log();
  console.log("- curr_date_code, curr_h, curr_min - ");
  console.log(curr_date_code);
  console.log(add_zero(curr_date_obj.getUTCHours()));
  console.log(add_zero(curr_date_obj.getUTCMinutes()));

    // var
    var msg_obj = {};

    // Match up day
  if(day_code == curr_date_code) {
    console.log();
    console.log(".. match up day ..");

    // Match up hour
    var curr_hour = add_zero(curr_date_obj.getUTCHours());
    if(schedule_hour == curr_hour) {
      console.log();
      console.log("~~ match up hour ~~");

      // Match up min
      var curr_min = add_zero(curr_date_obj.getUTCMinutes());
      if(schedule_min == curr_min) {
        console.log();
        console.log("## match up d-h-m, run ##");

                msg_obj = {
                    schedule_id: schedule_id,
                    device_name: device_name,
                    cmd: cmd,
                    is_run: true                                    
                };

                return Promise.resolve(msg_obj);            
      }
    }
  }
  else {

  }

    //
    msg_obj = {
    schedule_id: schedule_id,
    device_name: device_name,
    cmd: cmd,
    is_run: false 
  };

    return Promise.resolve(msg_obj);
}


// NOTE -------------
function process_schedule_rows(rows) {
    return Promise.mapSeries(rows, function(row) {
        return is_matchup_schedule_time(row)
            .then(is_schedule_cmd_already_pending)
            .then(insert_into_pending_cmd)
            .catch(function(e){
                throw e;
            })
    });
}


function do_schedule() {
  console.log();
  console.log("---- start do_schedule ----");

  g_pool.query("select * from Schedule order by ScheduleId asc")
  .then(process_schedule_rows)
    .catch(function(e){
        throw e;
    });
}


// main func
function main() {
    console.log("db host:");
  console.log(config.db_host);

    connect_db();

    interval(function(){
        do_schedule();
    }, 5000, undefined);

    // Clean up
  ON_DEATH(function(signal, err) {
    console.log();
    console.log("-- script interupted --");
    console.log("close db");

    // close db
    close_db();    

    process.exit();
  });

}


// run main func
main();

update 1

我已将代码更新为递归，但插入仍然发生两次。

function do_schedule() {
  console.log();
  console.log("---- start do_schedule ----");

  g_pool.query("select * from Schedule order by ScheduleId asc")
  .then(process_schedule_rows)
  .delay(2000)
  .then(do_schedule)
  .catch(function(e){
    throw e;
  });
}

update 2

当前，该脚本正在检查命令是否已插入数据库，如果是，则转到下一个任务。如果不是，它将命令插入到挂起表中。

该脚本每2秒检查一次，对于mysql检查是否已经有命令来说可能太快了。

嗯，我把它设置为2分钟，同样的问题仍然出现。 i.g 插入两次。

function do_schedule() {
  console.log();
  console.log("---- start do_schedule ----");

  g_pool.query("select * from Schedule order by ScheduleId asc")
  .then(process_schedule_rows)
  .delay(120000) <------------
  .then(do_schedule)
  .catch(function(e){
    throw e;
  });
}