基于大数据日志系统的流量回放平台

CREATE TABLE tbl_quartz_job (  id int unsigned NOT NULL AUTO_INCREMENT COMMENT '自增主键',  job_name varchar(50) NOT NULL COMMENT '任务名称',  job_group varchar(50) NOT NULL COMMENT '任务分组',  job_class_name varchar(100) NOT NULL COMMENT '执行类',  cron_expression varchar(100) NOT NULL COMMENT 'cron表达式',  trigger_state varchar(15) NOT NULL  COMMENT '任务状态',  old_job_name varchar(50) NOT NULL DEFAULT '' COMMENT '修改之前的任务名称',  old_job_group varchar(50) NOT NULL DEFAULT '' COMMENT '修改之前的任务分组',  description varchar(100) NOT NULL COMMENT '描述',  PRIMARY KEY (id),  UNIQUE KEY un_group_name (job_group,job_name)) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='定时任务';

     public boolean saveJob(QuartzJob job) {        int res;        String jobName = job.getJobName();        String jobGroup = job.getJobGroup();        try {            schedulerJob(job);            job.setTriggerState(JobStatus.RUNNING.getStatus());            job.setOldJobGroup(jobGroup);            job.setOldJobName(jobName);            res = jobMapper.saveJob(job);        } catch (Exception e) {            LOG.error("Failed to save job: ", e);            return false;        }
        if (res > 0) {            LOG.info("Success to save job. jobName: {}, jobGroup: {}", jobName, jobGroup);            return true;        }        return false; }
public void schedulerJob(QuartzJob job) throws Exception {        // job里需要指定具体的执行类(全路径)        Class cls = Class.forName(job.getJobClassName());
        // JobDetail表示一个具体的可执行的调度程序，Job是这个可执行程调度程序所要执行的内容，另外JobDetail还包含了这个任务调度的方案和策略。        JobDetail jobDetail = JobBuilder.newJob(cls).withIdentity(job.getJobName(), job.getJobGroup())                .withDescription(job.getDescription()).build();                // Scheduler有simpleScheduleBuilder和cronScheduleBuilder，这里是用cron方式声明定时任务的调度        CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression().trim());
        // Trigger代表一个调度参数的配置，什么时候去调        Trigger trigger = TriggerBuilder.newTrigger().withIdentity(TRIGGER_IDENTITY + job.getJobName(), job.getJobGroup())                .startNow().withSchedule(cronScheduleBuilder).build();                // Scheduler代表一个调度容器，一个调度容器中可以注册多个JobDetail和Trigger。当Trigger与JobDetail组合，就可以被Scheduler容器调度了。        scheduler.scheduleJob(jobDetail, trigger);    }

@Componentpublic class ApplicationInit implements CommandLineRunner {
  @Autowired    private JobService jobService;
    @Autowired    private Scheduler scheduler;
    @Override    public void run(String... args) throws Exception {        loadJobToQuartz();    }
    private void loadJobToQuartz() throws Exception {        List<QuartzJob> jobs = jobService.listQuartzJob();        for(QuartzJob job : jobs) {            jobService.schedulerJob(job);            if (JobStatus.PAUSED.getStatus().equals(job.getTriggerState())) {                scheduler.pauseJob(new JobKey(job.getJobName(), job.getJobGroup()));            }        }    }}

// 由于Kafka非线程安全，禁止并发执行@DisallowConcurrentExecutionpublic class BizdcRepeatJob extends QuartzJobBean {
    @Autowired    private BizdcRepeaterComponent component;
    @Override    protected void executeInternal(JobExecutionContext jobExecutionContext) {        component.repeat();    }}

private CompareResult sendRequestAndDiff(final String url1, final String url2, final String body) {        return CompletableFuture.supplyAsync(                () -> sendRequestAsync(url1, body), executorService        ).thenCombine(                CompletableFuture.supplyAsync(                        () -> sendRequestAsync(url2, body), executorService                ), (response1, response2) ->                        DiffUtils.commonDiff(response1, response2)        ).join();    }