Periodic jobs when running multiple servers

Question

I plan to deploy the application using Play and have never used their “Jobs” before. My deployment will be large enough to require different game servers to be load balanced, but my calculations will not be large enough to use hadoop / storm / others.

My question is: how do I handle this script on Play? If I set up work on Play every minute, I don’t want every server to do the same thing at the same time.

I could find this answer , but I don't like any of these options.

So, are there any tools or best practices for coordinating jobs or do I need to do something from scratch?

Answer 1

You can use the database flag as described here: Collaborate Pere Villega Playframework for two jobs.

But I think the solution from Guillaume Bort in Google Groups to use Memcache is the best. There seems to be a module for Play 2: https://github.com/mumoshu/play2-memcached

Answer 2

I would personally use one instance that performs tasks just for simplicity. Alternatively, you can use Akka instead of Job if you want finer control over execution and better concurrency, parallel processing.

Answer 3

You can use the table in your database to store jobLock, but you should check / update this lock in separate transactions (for this you need to use JPA.newEntityManager)

My JobLock class uses the LockMode enumeration

package enums; public enum LockMode { FREE, ACQUIRED; } 

here is the class JobLock

 package models; import java.util.Date; import java.util.List; import javax.persistence.Entity; import javax.persistence.EntityManager; import javax.persistence.EnumType; import javax.persistence.Enumerated; import javax.persistence.Version; import play.Logger; import play.Play; import play.data.validation.Required; import play.db.jpa.JPA; import play.db.jpa.Model; import utils.Parser; import enums.LockMode; import exceptions.ServiceException; /** * Technical class that allows to manage a lock in the database thus we can * synchronize multiple instances that thus cannot run the same job at the same * time * * @author sebastien */ @Entity public class JobLock extends Model { private static final Long MAX_ACQUISITION_DELAY = Parser.parseLong(Play.configuration.getProperty( "job.lock.acquisitiondelay", "10000")); @Required public String jobName; public Date acquisitionDate; @Required @Enumerated(EnumType.STRING) public LockMode lockMode; @Version public int version; // STATIC METHODS // ~~~~~~~~~~~~~~~ /** * Acquire the lock for the type of job identified by the name parameter. * Acquisition of the lock is done on a separate transaction thus is * transaction is as small as possible and other instances will see the lock * acquisition sooner. * <p> * If we do not do that, the other instances will be blocked until the * instance that acquired the lock have finished is businees transaction * which could be long on a job. * </p> * * @param name * the name that identifies a job category, usually it is the job * simple class name * @return the lock object if the acquisition is successfull, null otherwise */ public static JobLock acquireLock(String name) { EntityManager em = JPA.newEntityManager(); try { em.getTransaction().begin(); List<JobLock> locks = em.createQuery("from JobLock where jobName=:name", JobLock.class) .setParameter("name", name).setMaxResults(1).getResultList(); JobLock lock = locks != null && !locks.isEmpty() ? locks.get(0) : null; if (lock == null) { lock = new JobLock(); lock.jobName = name; lock.acquisitionDate = new Date(); lock.lockMode = LockMode.ACQUIRED; em.persist(lock); } else { if (LockMode.ACQUIRED.equals(lock.lockMode)) { if ((System.currentTimeMillis() - lock.acquisitionDate.getTime()) > MAX_ACQUISITION_DELAY) { throw new ServiceException(String.format( "Lock is held for too much time : there is a problem with job %s", name)); } return null; } lock.lockMode = LockMode.ACQUIRED; lock.acquisitionDate = new Date(); lock.willBeSaved = true; } em.flush(); em.getTransaction().commit(); return lock; } catch (Exception e) { // Do not log exception here because it is normal to have exception // in case of multi-node installation, this is the way to avoid // multiple job execution if (em.getTransaction().isActive()) { em.getTransaction().rollback(); } // Maybe we have to inverse the test and to define which exception // is not problematic : exception that denotes concurrency in the // database are normal if (e instanceof ServiceException) { throw (ServiceException) e; } else { return null; } } finally { if (em.isOpen()) { em.close(); } } } /** * Release the lock on the database thus another instance can take it. This * action change the {@link #lockMode} and set {@link #acquisitionDate} to * null. This is done in a separate transaction that can have visibility on * what happens on the database during the time of the business transaction * * @param lock * the lock to release * @return true if we managed to relase the lock and false otherwise */ public static boolean releaseLock(JobLock lock) { EntityManager em = JPA.newEntityManager(); if (lock == null || LockMode.FREE.equals(lock.lockMode)) { return false; } try { em.getTransaction().begin(); lock = em.find(JobLock.class, lock.id); lock.lockMode = LockMode.FREE; lock.acquisitionDate = null; lock.willBeSaved = true; em.persist(lock); em.flush(); em.getTransaction().commit(); return true; } catch (Exception e) { if (em.getTransaction().isActive()) { em.getTransaction().rollback(); } Logger.error(e, "Error during commit of lock release"); return false; } finally { if (em.isOpen()) { em.close(); } } } } 

and here is my LockAwareJob that uses this lock

 package jobs; import models.JobLock; import notifiers.ExceptionMails; import play.Logger; import play.jobs.Job; public abstract class LockAwareJob<V> extends Job<V> { @Override public final void doJob() throws Exception { String name = this.getClass().getSimpleName(); try { JobLock lock = JobLock.acquireLock(name); if (lock != null) { Logger.info("Starting %s", name); try { doJobWithLock(lock); } finally { if (!JobLock.releaseLock(lock)) { Logger.error("Lock acquired but cannot be released for %s", name); } Logger.info("End of %s", name); } } else { Logger.info("Another node is running %s : nothing to do", name); } } catch (Exception ex) { ExceptionMails.exception(ex, String.format("Error while executing job %s", name)); throw ex; } } @Override public final V doJobWithResult() throws Exception { String name = this.getClass().getSimpleName(); try { JobLock lock = JobLock.acquireLock(name); if (lock != null) { Logger.info("Starting %s", name); try { return resultWithLock(lock); } finally { if (!JobLock.releaseLock(lock)) { Logger.error("Lock acquired but cannot be released for %s", name); } Logger.info("End of %s", name); } } else { Logger.info("Another node is running %s : nothing to do", name); return resultWithoutLock(); } } catch (Exception ex) { ExceptionMails.exception(ex, String.format("Error while executing job %s", name)); throw ex; } } public void doJobWithLock(JobLock lock) throws Exception { } public V resultWithLock(JobLock lock) throws Exception { doJobWithLock(lock); return null; } public V resultWithoutLock() throws Exception { return null; } } 

In my log4j.properties I add a special line to avoid an error every time the instance was unable to get the job lock

 log4j.logger.org.hibernate.event.def.AbstractFlushingEventListener=FATAL 

With this solution, you can also use the JobLock ID to store the parameters associated with this job (e.g. last run)