Spring scheduler Task

[Baskar Sikkayan]

Hi,
I use Spring Quartz scheduler(Clustered) in my application as follows. Inside the scheduler, i instantiate a thread class to spawn a new thread for each record and inside the thread class, i use reference of other bean class using wiring concept. As the thread class inside the scheduler is instantiated using the new operator, the dependency injection inside the thread class is not working. Is there any way, we can achieve this?

From applicationContext.xml,

    <bean id="campaignNowJob" class="com.nervytech.mailer24x7.campaign.job.scheduler.CampaignNowJob"/>

    <bean id="campaignNowJobDetail" class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
       <property name="targetObject" ref="campaignNowJob" />
       <property name="targetMethod" value="run" />
       <property name="concurrent" value="false" />
    </bean>

        <bean id="campaignNowTrigger" class="org.springframework.scheduling.quartz.SimpleTriggerBean">	
	  <property name="jobDetail" ref="campaignNowJobDetail" />	
	  <property name="startDelay" value="6000" />
	  <!-- repeat every 90 seconds -->
	  <property name="repeatInterval" value="90000"/>
      </bean>

       <bean class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
	   <property name="triggers">
		 <list>
	  	    <ref bean="campaignNowTrigger" /> 
		 </list>
	   </property>

CampaignNow.java class,

/**
 * 
 */
package com.nervytech.mailer24x7.campaign.job.scheduler;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;

import com.nervytech.mailer24x7.campaign.job.task.CampaignTask;
import com.nervytech.mailer24x7.client.exception.MailerException;
import com.nervytech.mailer24x7.common.enums.CampaignStatusEnum;
import com.nervytech.mailer24x7.common.util.MailerUtil;
import com.nervytech.mailer24x7.model.dao.CampaignDAO;
import com.nervytech.mailer24x7.model.domains.Campaign;

/**
 * @author bsikkaya
 * 
 */
public class CampaignNowJob {
	
	private static final Logger logger = LoggerFactory
			.getLogger(CampaignNowJob.class);

	@Autowired
	private CampaignDAO campaignDAO;
	
	public void run() {
		
		logger.info("CampaignNow Job started !!!!");
		
		try {
			
			ExecutorService exec = Executors
					.newFixedThreadPool(MailerUtil.THREADS_COUNT);
			ThreadPoolExecutor executor = (ThreadPoolExecutor) exec;
			
			List<Campaign> campaignsList = .....;

			for (Campaign cmpn : campaignsList) {
				
				[b]CampaignTask task = new CampaignTask(Thread.currentThread().getName(), cmpn,
						content);
				[/b]
				try {
					executor .execute(task);
				} catch (RejectedExecutionException e) {
					// will be thrown if rejected execution handler
					// is not set with executor.setRejectedExecutionHandler
					e.printStackTrace();
				}
				
				logger.info("Campaign sent successfully for the campaignId : "+cmpn.getCampaignId());
				
				System.out.println("SUCCESSSSSSSSSSSSSSSSSSSSSs ");
			}

			shutDownAndWaitToFinish(executor);
			
			logger.info("CampaignNow Job Finished !!!!");
			System.out.println("FINISHEDDDDDDDDDDDDDDDDDDd");

		} catch (Exception e) {
			---------------
		}
	}
	
	public static void shutDownAndWaitToFinish(ThreadPoolExecutor threadPool)throws Exception
	{
		// Code to shutdown .........
	}

}

Thread class,

/**
 * 
 */
package com.nervytech.mailer24x7.campaign.job.task;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.nervytech.mailer24x7.common.enums.CampaignStatusEnum;
import com.nervytech.mailer24x7.common.util.MailerUtil;
import com.nervytech.mailer24x7.mailgun.MailgunCampignSyncer;
import com.nervytech.mailer24x7.mailgun.MailgunSendUtil;
import com.nervytech.mailer24x7.model.dao.SubscriberDAO;
import com.nervytech.mailer24x7.model.domains.Campaign;

/**
 * @author bsikkaya
 * 
 */

public class CampaignTask implements Runnable {

	private static final Logger logger = LoggerFactory
			.getLogger(CampaignTask.class);

	@Autowired
	private SubscriberDAO subscriberDAO;

	@Autowired
	private MailgunCampignSyncer mailgunCampignSyncer;

	private SimpleDateFormat fmt = new SimpleDateFormat(
			"yyyy-MM-dd hh:mm:ss.SSS");
	private String name;
	private Date created;
	private Campaign cmpn;
	private String content;
	

	public CampaignTask(String name, Campaign cmpn, String content) {
		this.name = name;
		this.cmpn = cmpn;
		this.content = content;
		this.created = new Date();
	}

	@Override
	public void run() {

		try {			

			mailgunCampignSyncer.updateCampaignInMailgun(cmpn);
			
			MailgunSendUtil.sendMail(cmpn.getCampaignId(), cmpn.getSenderEmailId(), cmpn.getSubscriberListId(), cmpn.getOrgId(), cmpn.getSubject(), null, content, null);

			subscriberDAO.updateCampaignStatus(cmpn.getCampaignId(), CampaignStatusEnum.COMPLETED.getStatus(), MailerUtil.formatter.format(new Date()));

		} catch (Exception e) {
			logger.error("Error sending campaign to :" + cmpn.getCampaignId()
					+ " ", e);
			e.printStackTrace();
		}
	}

	@Override
	public String toString() {
		return name + ", created " + fmt.format(created);
	}

}

Thanks,
Baskar.S

[Bill Gorder]

There are a couple of ways you could handle this. My first suggestion would be to make an async service bean.

public class AsyncCampaignServiceImpl implements AsyncCamapignService{

    @Autowired  
    private SubscriberDAO subscriberDAO;  
  
    @Autowired  
    private MailgunCampignSyncer mailgunCampignSyncer;  

    @Override
    @Async
    public void processCampaign(String name, Campaign cmpn, String content){
        //do async stuff here 
    }
}

Then autowire AsyncCamapignService in your CampaignNowJob and call the method.

dont forget to add

<task:annotation-driven"/>

to your XML configuration.

see the docs for more info on @Async
http://static.springsource.org/spring/docs/3.1.x/spring-framework-reference/html/scheduling.html#scheduling-annotation-support-async

[Baskar Sikkayan]

Hi,
Thanks for the info.

I may not be able to go with this as i use ThreadPoolExecutor and the excute method needs runnable implementor. Also, i can configure many things in ThreadPoolExecutor. Is there any other equivalent class in Spring?

Thanks,
Baskar.S

[Bill Gorder]

You can specify an executor on the annotation-driven tag. Alternatively you can specify any executor on the annotation itself see the documentation here:
http://static.springsource.org/spring/docs/3.1.x/spring-framework-reference/html/scheduling.html#scheduling-annotation-support-qualification

The executor provided must of course must have a bean definition in your context. It can be any Executor of type java.util.concurrent.Executor or org.springframework.core.task.TaskExecutor. Since ThreadPoolExecutor meets this criteria there is no reason you could not use it. Be aware if there are multiple methods within the class needing the async behavior this annotation can be applied at the class level as well.