spring - 如何使长时间运行的@Scheduled spring 方法， hibernate 工作？

Question

我尝试制作一个 Jersey 网络服务，允许客户创建工作。这些作业存储在数据库中，使用 Hibernate 作为持久性提供程序。这些作业将由计划服务在后台执行，我想用 Spring 来安排。

我创建了一个 Spring Scheduled 方法，如下所示:

@Service
public class MyTimedService
{
    @Inject
    IJobs allJobs;

    private static final Logger LOG = LoggerFactory.getLogger( MyTimedService.class );


    @Scheduled(fixedRate=5000)
    public void processJobs()
    {
        for(BaseJob job: allJobs.getQueuedJobs())
        {
            processJob(job, new JobContext());
        }
    }


private void processJob( final BaseJob job, JobContext context ) throws JobException
{
    job.start();

    LOG.info( "Starting: " + job.getName() );
    job.execute( context );
    LOG.info( "Finished: " + job.getName() );

    if ( job.getErrors().size() > 0 )
    {
        Throwable e = job.getErrors().get( 0 );
        throw new JobException( e );
    }
    job.finished();

}
...
}

因为 Job 会运行很长时间，所以我需要让 job.start() 向数据库报告状态变化(从 QUEUE 到 IN_PROGRESS)。之前我用的是命令行实现，有自己的事务管理，基本上是 begin()和 commit()就在 job.start()附近.

现在我需要使用 Spring 使其工作......

关于如何分离关注点并使其发挥作用的任何建议？

Answer 1

编辑

One thing I do not really understand is why the doWork needs one big transaction.

它不一定是。任何一个方向都有警告。我在 doWork(...) 方法上方的修改后的类打击 (JobRunnerService) 中注意到了其中的一些。这些笔记是值得注意的。

What I would like to achive is that the doWork regularly can set the job's progress

这可能很难实现，也可能不难实现，具体取决于您是否希望 doWork(...) 绑定(bind)到 Transaction 以及每个 Job 是否可以以相同的方式分解(即:更新总是发生在静态代码中的位置)。我不知道您的所有要求，所以我无法真正回答这个问题。但是，我会重申我对研究 Spring Batch 的建议。

JobRunnerService

import me.mike.jobs.model.Job;
import me.mike.jobs.model.JobState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;

/**
 * !!This bean is STATEFUL!!
 */
@Service
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class JobRunnerService {
    @Autowired
    private JobService js;

    public void processJob(Job job) {
        job.setState(JobState.WORKING_0);
        js.update(job);
        try {
            doWork(job);
            job.setState(JobState.COMPLETE);
        } catch (Exception e) {
            job.setState(JobState.FAILED);
        }
        System.out.println("I'm done working.");
        js.update(job);
    }

    /**
     * Be sure that any unchecked exception you throw gets added into the "rollbackFor" since it won't trigger
     * a rollback if you don't...
     *
     * The @Transactional is optional - I assumed you would want the work performed in the job to be transactional.
     *
     * Note: Remember, when doing the work represented by these jobs, that your EntityManager (or SessionFactory) is
     * configured with a TransactionManager and, as such, will throw exceptions when you attempt to do work within them
     * without a Transaction.  You will either need a separate EntityManager (SessionFactory) or something like a
     * JdbcTemplate.
     *
     * Note: If the Job's work DOES need to be Transactional, this will probably not work.  A very simple solution
     * would to be to split up the work within the job into "steps" or "stages."  The processJob(...) method above
     * could then call each stage and, at the conclusion, update the Job's state appropriately.  This, of course,
     * would not work if each Job had N number of stages where N could vary an indeterminate amount.
     */
    //@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = { IllegalArgumentException.class })
    public void doWork(Job job) throws IllegalArgumentException {
        // This method begins its own transaction, every single time its called.  Period.
        // Do some work...
        job.setState(JobState.WORKING_10);
        js.update(job);
        // Do more work...
        job.setState(JobState.WORKING_90);
        js.update(job);
        // At the conclusion, the transaction bound to this method is committed, unless a rollback was initiated.
    }
}

前言:
我认为你最好考虑利用像 SpringBatch 这样的东西。它可能需要更多的配置，但它也提供了更多的支持。

如果我理解正确，您希望将“作业”存储在表中(RESTful 创建)。您需要一个可以在后台定期运行的 @Scheduled 任务来执行每个作业所代表的工作。您还希望在处理这些实体之前和之后更改每个实体的状态 (heh)。需要注意的是，初始状态更改需要在其自己的事务范围内发生，不可避免的结束状态更改也是如此。

我已经使用 Spring、JPA 和 Hibernate 对 MySQL 5.x DB 运行了此代码。如果您需要，我可以为您提供我的 applicationContext 和我的 rest-servlet xml 文件。

这将执行我理解的您的既定目标:

型号:

import org.hibernate.validator.constraints.Length;

import javax.persistence.*;
import javax.validation.constraints.NotNull;
import java.util.UUID;

@Entity
public class Job {
    @Id
    private String id;

    @Column
    @NotNull
    @Length(min = 3, max = 50)
    private String name;

    @Enumerated(EnumType.STRING)
    @Column(length = 50, nullable = false)
    private JobState state;

    public UUID getId() {
        return UUID.fromString(id);
    }

    public void setId(UUID id) {
        this.id = id.toString();
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public JobState getState() {
        return state;
    }

    public void setState(JobState state) {
        this.state = state;
    }
}

存储库:

import me.mike.jobs.model.Job;
import me.mike.jobs.model.JobState;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.persistence.criteria.CriteriaBuilder;
import javax.persistence.criteria.CriteriaQuery;
import javax.persistence.criteria.Root;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

@Repository
public class JobDao {
    @PersistenceContext
    private EntityManager em;


    @Transactional(propagation = Propagation.REQUIRED)
    public void create(Job job) {
        // ...
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public Set<Job> readAll() {
        // ...
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public Job readById(UUID id) {
        // ...
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public Set<Job> readByState(JobState state) {
        // ...
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void update(Job job) {
        // ...
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void delete(Job job) {
        // ...
    }
}

工作服务 (这将处理您的 Job 实体上的 RESTful 操作)

import me.mike.jobs.dao.JobDao;
import me.mike.jobs.model.Job;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Set;

@Service
public class JobService {
    @Autowired
    private JobDao jd;

    @Transactional(propagation = Propagation.REQUIRED)
    public void create(Job job) {
        // Business logic...
        jd.create(job);
        // More business logic...
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public Set<Job> read() {
        // Business logic...
        Set<Job> jobs = jd.readAll();
        // More business logic...
        return jobs;
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void update(Job job) {
        // Business logic...
        jd.update(job);
        // More business logic...
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void delete(Job job) {
        // Business logic...
        jd.delete(job);
        // More business logic...
    }
}

维修服务 (这个人会持有你所有的 @ScheduledTask 方法)

import me.mike.jobs.dao.JobDao;
import me.mike.jobs.model.Job;
import me.mike.jobs.model.JobState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

@Service
public class MaintenanceService {
    @Autowired
    private JobRunnerService jrs;

    @Autowired
    private JobDao jd;

    @Scheduled(fixedDelay = 5000, initialDelay = 5000)
    public void processQueuedJobs() {
        // This may be somewhat dangerous depending on how many jobs could potentially be racked up during the 'downtime'
        for (Job curJob : jd.readByState(JobState.QUEUED))
            jrs.processJob(curJob);
    }

    // Any other timed service methods...
}

JobRunnerService 这是实际运行作业的服务

import me.mike.jobs.model.Job;
import me.mike.jobs.model.JobState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

/**
 * !!This bean is STATEFUL!!
 */
@Service
@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class JobRunnerService {
    @Autowired
    private JobService js;

    public void processJob(Job job) {
        job.setState(JobState.WORKING);
        js.update(job);
        try {
            doWork(job);
            job.setState(JobState.COMPLETE);
        } catch (Exception e) {
            job.setState(JobState.FAILED);
        }
        System.out.println("I'm done working.");
        js.update(job);
    }

    /**
     * Be sure that any unchecked exception you throw gets added into the "rollbackFor" since it won't trigger
     * a rollback if you don't...
     */
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = { IllegalArgumentException.class })
    public void doWork(Job job) throws IllegalArgumentException {
        // This method begins its own transaction, every single time its called.  Period.
        // Do your work here...
        // At the conclusion, the transaction bound to this method is committed, unless a rollback was initiated.
    }
}