CBSNexora/cbsnexora-backend
4
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

added scheduler groups for dependent jobs

Browse Source
This commit is contained in:
Pramod Nuthakki
2013-07-24 09:21:51 +05:30
parent 8d58cb8a84
commit 5fd83722ed
7 changed files with 163 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mifosng-db/migrations/core_db/V89__added_scheduler_group.sql
Executable
+2
View File
@@ -0,0 +1,2 @@
ALTER TABLE `job`
ADD COLUMN `scheduler_group` SMALLINT(2) NOT NULL DEFAULT '0' AFTER `updates_allowed`;
mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java
+11
View File
@@ -62,6 +62,13 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
@Column(name = "currently_running")
private boolean currentlyRunning;
@SuppressWarnings("unused")
@Column(name = "updates_allowed")
private boolean updatesAllowed;
@Column(name = "scheduler_group")
private Short schedulerGroup;
protected ScheduledJobDetail() {
}
@@ -86,6 +93,10 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
return this.jobKey;
}
public Short getSchedulerGroup() {
return this.schedulerGroup;
}
public boolean isActiveSchedular() {
return this.activeSchedular;
}
mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java
+6
View File
@@ -6,4 +6,10 @@ public interface JobRegisterService {
public void rescheduleJob(Long jobId);
public void stopScheduler();
public void startScheduler();
public boolean isSchedulerRunning();
}
mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java
+139 -27
View File
@@ -1,6 +1,9 @@
package org.mifosplatform.infrastructure.jobs.service;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.TimeZone;
import javax.annotation.PostConstruct;
@@ -14,6 +17,8 @@ import org.mifosplatform.infrastructure.security.service.TenantDetailsService;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.JobListener;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.slf4j.Logger;
@@ -38,18 +43,19 @@ public class JobRegisterServiceImpl implements JobRegisterService {
private final ApplicationContext applicationContext;
private final SchedulerFactoryBean schedulerFactoryBean;
private final SchedularWritePlatformService schedularWritePlatformService;
private final TenantDetailsService tenantDetailsService;
private final HashMap<Long, Boolean> schedulerStatus = new HashMap<Long, Boolean>(2);
private final HashMap<String, Scheduler> schedulers = new HashMap<String, Scheduler>(4);
@Autowired
public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedulerFactoryBean schedulerFactoryBean,
final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) {
public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedularWritePlatformService schedularService,
final TenantDetailsService tenantDetailsService) {
this.applicationContext = applicationContext;
this.schedularWritePlatformService = schedularService;
this.schedulerFactoryBean = schedulerFactoryBean;
this.tenantDetailsService = tenantDetailsService;
}
@@ -59,6 +65,7 @@ public class JobRegisterServiceImpl implements JobRegisterService {
for (MifosPlatformTenant tenant : allTenants) {
ThreadLocalContextUtil.setTenant(tenant);
List<ScheduledJobDetail> scheduledJobDetails = this.schedularWritePlatformService.retrieveAllJobs();
schedulerStatus.put(tenant.getId(), true);
for (ScheduledJobDetail jobDetails : scheduledJobDetails) {
scheduleJob(jobDetails);
this.schedularWritePlatformService.saveOrUpdate(jobDetails);
@@ -66,14 +73,18 @@ public class JobRegisterServiceImpl implements JobRegisterService {
}
}
public void executeJob(final JobKey jobKey) {
public void executeJob(final ScheduledJobDetail scheduledJobDetail) {
try {
final JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE, SchedularServiceConstants.TRIGGER_TYPE_APPLICATION);
jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier());
if (schedulerFactoryBean.getScheduler().checkExists(jobKey)) {
String key = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(key);
String schedulerName = getSchedulerName(scheduledJobDetail);
Scheduler scheduler = schedulers.get(schedulerName);
if (scheduler != null && scheduler.checkExists(jobKey)) {
try {
schedulerFactoryBean.getScheduler().triggerJob(jobKey, jobDataMap);
scheduler.triggerJob(jobKey, jobDataMap);
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
@@ -86,17 +97,51 @@ public class JobRegisterServiceImpl implements JobRegisterService {
}
public void rescheduleJob(ScheduledJobDetail scheduledJobDetails) {
public void rescheduleJob(ScheduledJobDetail scheduledJobDetail) {
try {
String jobIdentity = scheduledJobDetails.getJobKey();
String jobIdentity = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(jobIdentity);
schedulerFactoryBean.getScheduler().deleteJob(jobKey);
scheduleJob(scheduledJobDetails);
this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails);
String schedulername = getSchedulerName(scheduledJobDetail);
Scheduler scheduler = schedulers.get(schedulername);
if (scheduler != null) {
scheduler.deleteJob(jobKey);
}
scheduleJob(scheduledJobDetail);
this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail);
} catch (Throwable throwable) {
String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.updateErrorLog(stackTrace);
this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails);
scheduledJobDetail.updateErrorLog(stackTrace);
this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail);
}
}
@Override
public void stopScheduler() {
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false);
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
try {
String schedulerName = schedulerEntry.getValue().getSchedulerName();
if (isCurrentTenantsScheduler(schedulerName)) {
schedulerEntry.getValue().standby();
}
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
}
}
@Override
public void startScheduler() {
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true);
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
try {
String schedulerName = schedulerEntry.getValue().getSchedulerName();
if (isCurrentTenantsScheduler(schedulerName)) {
schedulerEntry.getValue().start();
}
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
}
}
@@ -111,9 +156,7 @@ public class JobRegisterServiceImpl implements JobRegisterService {
try {
ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId);
if (scheduledJobDetail != null) {
String key = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(key);
executeJob(jobKey);
executeJob(scheduledJobDetail);
} else {
throw new JobNotFoundException(String.valueOf(jobId));
}
@@ -122,6 +165,22 @@ public class JobRegisterServiceImpl implements JobRegisterService {
}
}
@Override
public boolean isSchedulerRunning() {
return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId());
}
private boolean isCurrentTenantsScheduler(String schedulerName) {
int beginIndex = SchedularServiceConstants.SCHEDULER.length();
int endIndex = schedulerName.length();
if (schedulerName.contains(SchedularServiceConstants.SCHEDULER_GROUP)) {
endIndex = schedulerName.indexOf(SchedularServiceConstants.SCHEDULER_GROUP);
}
String tenantId = schedulerName.substring(beginIndex, endIndex);
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
return tenant.getId() == Long.valueOf(tenantId);
}
private void scheduleJob(ScheduledJobDetail scheduledJobDetails) {
if (!scheduledJobDetails.isActiveSchedular()) {
scheduledJobDetails.updateNextRunTime(null);
@@ -129,9 +188,10 @@ public class JobRegisterServiceImpl implements JobRegisterService {
return;
}
try {
Scheduler scheduler = getScheduler(scheduledJobDetails);
JobDetail jobDetail = createJobDetail(scheduledJobDetails);
Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
schedulerFactoryBean.getScheduler().scheduleJob(jobDetail, trigger);
scheduler.scheduleJob(jobDetail, trigger);
scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey()));
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
scheduledJobDetails.updateErrorLog(null);
@@ -143,17 +203,52 @@ public class JobRegisterServiceImpl implements JobRegisterService {
scheduledJobDetails.updateCurrentlyRunningStatus(false);
}
private Scheduler getScheduler(ScheduledJobDetail scheduledJobDetail) throws Exception {
String schedulername = getSchedulerName(scheduledJobDetail);
Scheduler scheduler = this.schedulers.get(schedulername);
if (scheduler == null) {
int noOfThreads = SchedularServiceConstants.DEFAULT_THREAD_COUNT;
if (scheduledJobDetail.getSchedulerGroup() > 0) {
noOfThreads = SchedularServiceConstants.GROUP_THREAD_COUNT;
}
scheduler = createScheduler(schedulername, noOfThreads);
this.schedulers.put(schedulername, scheduler);
}
return scheduler;
}
private String getSchedulerName(ScheduledJobDetail scheduledJobDetail) {
StringBuilder sb = new StringBuilder(20);
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
sb.append(SchedularServiceConstants.SCHEDULER).append(tenant.getId());
if (scheduledJobDetail.getSchedulerGroup() > 0) {
sb.append(SchedularServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup());
}
return sb.toString();
}
private Scheduler createScheduler(String name, int noOfThreads) throws Exception {
SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
schedulerFactoryBean.setSchedulerName(name);
schedulerFactoryBean.setGlobalJobListeners(getGlobalListener());
Properties quartzProperties = new Properties();
quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads));
schedulerFactoryBean.setQuartzProperties(quartzProperties);
schedulerFactoryBean.afterPropertiesSet();
schedulerFactoryBean.start();
return schedulerFactoryBean.getScheduler();
}
private JobListener[] getGlobalListener() throws ClassNotFoundException {
JobListener listener = (JobListener) getBeanObject(SchedulerJobListener.class);
JobListener[] listenerArray = { listener };
return listenerArray;
}
private JobDetail createJobDetail(ScheduledJobDetail scheduledJobDetails) throws Exception {
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetails.getJobName());
String[] beanNames = this.applicationContext.getBeanNamesForType(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX]));
Object targetObject = this.applicationContext.getBean(beanNames[0]);
for (String beanName : beanNames) {
Object nextObject = this.applicationContext.getBean(beanName);
if (nextObject != null && !targetObject.getClass().isAssignableFrom(nextObject.getClass())) {
targetObject = nextObject;
}
}
Object targetObject = getBeanObject(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX]));
MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail" + tenant.getId());
jobDetailFactoryBean.setTargetObject(targetObject);
@@ -164,6 +259,23 @@ public class JobRegisterServiceImpl implements JobRegisterService {
return jobDetailFactoryBean.getObject();
}
/**
* @param jobDetails
* @return
* @throws ClassNotFoundException
*/
private Object getBeanObject(Class<?> classType) throws ClassNotFoundException {
String[] beanNames = this.applicationContext.getBeanNamesForType(classType);
Object targetObject = this.applicationContext.getBean(beanNames[0]);
for (String beanName : beanNames) {
Object nextObject = this.applicationContext.getBean(beanName);
if (nextObject != null && !targetObject.getClass().isAssignableFrom(nextObject.getClass())) {
targetObject = nextObject;
}
}
return targetObject;
}
/**
* @param scheduledJobDetails
* @return
mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java
+5
View File
@@ -14,4 +14,9 @@ public interface SchedularServiceConstants {
public static final String DEFAULT_LISTENER_NAME = "Global Listner";
public static final int STACK_TRACE_LEVEL = 7;
public static final String TENANT_IDENTIFIER = "tenantIdentifier";
public static final String SCHEDULER = "Scheduler";
public static final String SCHEDULER_GROUP = "group";
public static final int DEFAULT_THREAD_COUNT = 7;
public static final int GROUP_THREAD_COUNT = 1;
}
mifosng-provider/src/main/resources/META-INF/spring/appContext.xml
-1
View File
@@ -51,5 +51,4 @@
<import resource="securityContext.xml" />
<import resource="scheduler.xml"/>
</beans>
mifosng-provider/src/main/resources/META-INF/spring/scheduler.xml
-22
View File
@@ -1,22 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
This Source Code Form is subject to the terms of the Mozilla Public
License, v. 2.0. If a copy of the MPL was not distributed with this file,
You can obtain one at http://mozilla.org/MPL/2.0/.
-->
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean name="schedulerFactory"
class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="globalJobListeners">
<list>
<ref bean = "schedulerJobListener"/>
</list>
</property>
</bean>
</beans>