quartz定时器在运营商的使用

Posted 2022-11-28 六楼外的风景

定时任务在长流程的业务中应该还是比较多的，一种是非实时接口文件接口，这类一般用shell的crontab定时执行脚本，但shell中处理复杂逻辑比较吃力，一般会放到java或c/c++中实现，用nohup后台运行命令启用一个调java的守护线程.

#!/bin/sh
ACTION="$1"
PROCNAME="$2"
EXE_USER=`whoami`
run_RECK=com.qyw.main.SchedulerStart
mem_min="128M"
mem_max="1024M"
perm_size="128m"
max_perm_size="256m"

#check args
if [ $# -ne 2 ]; then
echo "Please input args.";
echo "example: uipquartzjob start(stop) jzjktag(agent)";
exit 1;
fi

if [ "$ACTION" != "start" ] && [ "$ACTION" != "stop" ] ; then
  echo "args invalid! the first arg should be start|stop";
   echo "example: reck start";
  exit 1;
fi

#execute
runclass=$run_RECK;

JAVA_LANGUAGE="-Dfile.encoding=GBK -Ddefault.client.encoding=GBK -Duser.language=zh -Duser.region=CN"
JPDA_OPTS="-agentlib:jdwp=transport=dt_socket,address=8005,server=y,suspend=n"

ps -ef|grep MODE="$PROCNAME"|grep "$EXE_USER"|grep -v grep|awk 'print $2' > $PROCNAME.pid
pid_count=`wc -l "$PROCNAME.pid"|awk 'print $1'`

if [ "$ACTION" = "start" ] ; then
  if [ "$pid_count"  -gt 0 ] ; then

    echo "$PROCNAME  had already running! $pid_count";

  else
      . setEnv.sh;
 nohup java $JAVA_LANGUAGE -DAPP=$PROCNAME -DTHR_NUM=5 -DWORK_LOAD=100 -DMODE=$PROCNAME -DPROV_CODE=051 -DDB_NUM=4 -DUIP_HOME=$HOME -DCONFIGPATH=$HOME/file/uip_quartz/etc/ -server -Xms$mem_min -Xmx$mem_max -XX:PermSize=$perm_size -XX:MaxPermSize=$max_perm_size $runclass  $PROCNAME >$HOME/log/$PROCNAME.log 2>&1 &
    echo "$PROCNAME  start to running...";

  fi

elif [ "$ACTION" = "stop" ] ; then

  if [ "$pid_count"  -gt 0 ] ; then
    for pid in `cat $PROCNAME.pid`
      do
        kill -9 $pid ;
      done    
    echo "kill $PROCNAME $EXE_USER done!";

  else

    echo "$PROCNAME $EXE_USER do not exist!";

  fi

fi

rm -f ./$PROCNAME.pid

如果每一个定时器都用到crontab来配置，当定时器越来越多时，crontab配置就显得不好维护，另外关键逻辑还是在java或c++中实现的，这时用对成熟的quartz框架就相当实用和重要,本文件就不讲quartz的下载配置相关知识,自行google主要讲解它项目中的应该场景!

1. 应该场景异步接口批量停开机
   背景知识:
   运营商每到月底都要做信控批量停开机,也就是这一切都是在另外一个信控系统发起,依据用户欠费和用户的信用额度决定停机的,一停就是上百W的量;停开机又是一个业务流程比较长且还依赖外部服开系统,所以接入接口都设计为异步.先实时接口信控发起的单子,然后写表保存,再启动quartz定时器快速重新发起做业务逻辑处理;当然现实中考虑情况复杂多了(如同一号码同时发起了两笔停机和开机单要怎么处理…,失败重发问题)
   

了解背景后,我们就讲下内部quartz设置

1. 1.2
   程序设计概述
   

对应上面,在我们系统中每个表都要分库分区的,一个表会分4个库来存建,所以每个定时器Job分定时循环建4个Thread线程,对应也会启动4套线程池,去执行Runable中的逻辑处理方法,然后再用数据库线程池去更新每个Runable逻辑所处理的数据;
对应如下关系的类OrderReceiveSoNotifyJob.java, OrderReceiveSoNotifyTh.java和OrderReceiveSoSoap.java

1. 1.3程序详细设计:
   Job类: OrderReceiveSoNotifyJob.java

import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import com.qyw.thread.OrderReceiveSoNotifyTh;
import com.util.PathConfig;
public class OrderReceiveSoNotifyJob implements Job 
  private static int count = 0;
  static String dbNum = PathConfig.getConfig("orderReceiveSo.dbNum");//可配置执行几库数据(1,3,2,4)
  //线程池大小
  static ExecutorService executorService =      Executors.newFixedThreadPool(Integer.parseInt(PathConfig.getConfig("orderReceiveSo.threadPool")));
  protected static final Logger logger = Logger.getLogger(OrderReceiveSoNotifyJob.class);
  public void execute(JobExecutionContext context) throws JobExecutionException 
    String [] strs = null;
    if (dbNum!=null&&!dbNum.equals("")) 
        strs = dbNum.split(",");
        for (int i =0; i < strs.length; i++)
          new OrderReceiveSoNotifyTh(strs[i], executorService).start();
        
      else 
        logger.info("db number is null:" + strs);
      
    

Thread类OrderReceiveSoNotifyTh.java

public class OrderReceiveSoNotifyTh extends Thread

  private String hostId = "1";
  private ExecutorService executorReceive = null;
  public OrderReceiveSoNotifyTh(String hostId, ExecutorService executorService) 
    this.hostId = hostId;
    this.executorReceive = executorService;
  
  public void run()
  
      //省略数据连接查询结果代码
      ..........
      //最终每条数据在map中,一次查600条数据(sql中用ROWNUM<600限定)都装在list中
      for (Map map : list) 
        logger.info(map);
        //逻辑处理类OrderReceiveSoSoap 实现Runable接口
        OrderReceiveSoSoap so = new OrderReceiveSoSoap(map,hostId);
        executorReceive.submit(so);
      

Runable线程OrderReceiveSoSoap.java

代码过多就不贴了这个类就像金字塔最低层的工人(理解成码农也对),做一堆的苦力活!!
quartz_job.xml

cron-expression表达式配置,此处是每10秒执行一次,具体配置和linux中crontab是一致的
注:线程池大小设置和sql查多少条数据是成一定比例的,经反复测试,15个线程跑600条数据基本上都在1秒内完成!

到此一个完整的quartz定时器就配置完了,但上面的程序是有问题的!
问题A:重复跑数据
继承job这个定时器是到时就执行的,不管上一次是否已经执行完!这样就会产生上一次job还没执行完,数据库中标记状态还没来得级变更,下一个定时器就启动了,又取了上一次的正在执行的数据,就是取数重复杂;
问题B: 扩展定时器
如果一个定时器跑的数据数量是有限的,那再增加多个定时器时跑同样数据时如何保证数据能快速发出

Job还有一类为有状态的StatefulJob接口，如果我们需要在上一个作业执行完后，根据其执行结果再进行下次作业的执行，则需要实现此接口;这个可以解决问题A;
但问题又来了,如果上一个定时器因一系列问题(数据连接/异常数据)一直没跑,那下一个定时器始终无法执行,相当于没法及时或延时给用户停开机,想象一下一个用记发现自己被停机了,马上冲值开机,结果等了半天没开机,不投诉你才怪…

引用临时表table_tmp
好处有3个
1. 每次查数据都关联table_tmp,查出只有不在临时表中的数据才处理,每次查到的数据又批量(手工提交事务)插入临时表中,耗时一般都不会超过100毫秒

注:记得把事务设置为false再批量提交

这样做的好处是,首先不会取到重得数据,其次可以无限扩展加多个定时器,当数据越多时,就可以配置多个job;
当一个job有问题或某些不可预知的问题,下一个job到来时会强行执行没处理过的数据;只要每个job中是取数sql都关联临时表每个定时器都不会再取到重复数据!

2. 重发问题;
以前做重发要不就是运维手工更新状态重发,要不就是另外写个job定更新失败状态重发;但目前因为多了临时表,就好好利用临时表的作用;
○1关联临时表取数时,把初始状态和失败状态也一起和临时表关联,正常情况是不会把失败数据取出的,因为临时表有相关数据;
○2写多个job定时删除临时表数据,具体多长时间删除一次,这个和业务中多久需要重一次数据一致即可;删除临时表数据既可以加快sql查询效率又可以重发数据,一举两得;

3. 这个金字塔的并发量情况如何,能否快速处理批量数据
解决上面两个问题”重复数据”和 “重发问题”问题,用这两把利器就可以无限增加job数量,或增加sql取数数量同时平衡job之间的时间间隔是可以做到短时间发送大批量数据! 最终这个实时性,数据量就转为job时间间隔、线程池数和sql取数数量之间的平衡艺术了;这个只能从实际实践中寻找;

最后说下面quartz_job.xml的crontab时间配置的区别
△10 0/1 * * * ?
—>每分钟执行一次是整分钟执行,如12:01分执行,下次是12:02分准时执行
△215 0/1 * * * ?
—>每分钟执行一次是整分钟延时15秒执行, 如12:01:15分执行,下次是12:02:15分准时执行

这个可能就是为了下面这个场景设计的:
如多个job都做同一件事时,可以利用这个延时执行,每个job都延时都不一样;这个好处正面说不清,反证下你就知道了;
要是每个job都在同一秒去数据库取数,那相当于只有一个job能取到数据,先执行的先抢了工作,后面的都没工作了,所以错时用工还是很必要的!