JPA有條件的插入

Question

我有一個基於Java Spring的Web應用程序，我只想在表中沒有包含任何與新的「相似」（根據一些特定的，不相關的標準）的行時將記錄插入到表中行。

因爲這是一個多線程環境，所以我不能使用SELECT + INSERT兩步組合，因爲它會將我暴露給競爭條件。

幾年前，同樣的問題首次被提出並回答了here和here。不幸的是，這些問題只有一點點關注，所提供的答案不足以滿足我的需求。

這是我目前擁有的代碼，它不工作：

@Component("userActionsManager") 
@Transactional 
public class UserActionsManager implements UserActionsManagerInterface { 

    @PersistenceContext(unitName = "itsadDB") 
    private EntityManager manager; 

    @Resource(name = "databaseManager") 
    private DB db; 

    ... 

    @SuppressWarnings("unchecked") 
    @Override 
    @PreAuthorize("hasRole('ROLE_USER') && #username == authentication.name") 
    public String giveAnswer(String username, String courseCode, String missionName, String taskCode, String answer) { 
     ... 

     List<Submission> submissions = getAllCorrectSubmissions(newSubmission); 
     List<Result>  results  = getAllCorrectResults(result); 

     if (submissions.size() > 0 
     || results.size()  > 0) throw new SessionAuthenticationException("foo"); 

     manager.persist(newSubmission); 
     manager.persist(result); 

     submissions = getAllCorrectSubmissions(newSubmission); 
     results  = getAllCorrectResults(result); 

     for (Submission s : submissions) manager.lock(s, LockModeType.OPTIMISTIC_FORCE_INCREMENT); 
     for (Result  r : results ) manager.lock(r, LockModeType.OPTIMISTIC_FORCE_INCREMENT); 

     manager.flush(); 

     ... 
    } 

    @SuppressWarnings("unchecked") 
    private List<Submission> getAllCorrectSubmissions(Submission newSubmission) { 
     Query q = manager.createQuery("SELECT s FROM Submission AS s WHERE s.missionTask = ?1 AND s.course = ?2 AND s.user = ?3 AND s.correct = true"); 
     q.setParameter(1, newSubmission.getMissionTask()); 
     q.setParameter(2, newSubmission.getCourse()); 
     q.setParameter(3, newSubmission.getUser()); 
     return (List<Submission>) q.getResultList(); 
    } 

    @SuppressWarnings("unchecked") 
    private List<Result> getAllCorrectResults(Result result) { 
     Query q = manager.createQuery("SELECT r FROM Result AS r WHERE r.missionTask = ?1 AND r.course = ?2 AND r.user = ?3"); 
     q.setParameter(1, result.getMissionTask()); 
     q.setParameter(2, result.getCourse()); 
     q.setParameter(3, result.getUser()); 
     return (List<Result>) q.getResultList(); 
    } 

... 

} 

按照答案提供here我應該以某種方式使用OPTIMISTIC_FORCE_INCREMENT但它不工作。我懷疑提供的答案是錯誤的，所以我需要一個更好的答案。

編輯：

增加了更多的上下文相關的代碼。現在這個代碼仍然存在競爭條件。當我同時創建10個HTTP POST請求時，大約5行將被錯誤地插入。其他5個請求被HTTP錯誤代碼409拒絕（衝突）。正確的代碼將保證只有1行將被插入到數據庫中，而不管我做出多少個併發請求。使方法同步不是一個解決方案，因爲競爭條件仍然表現出某種未知的原因（我測試過）。

Answer 1

不幸的是經過幾天的研究，我無法找到一個簡短而簡單的解決方案來解決我的問題。由於我的時間預算不是無限的，我不得不想出一個解決方法。如果你願意的話，可以把它叫做kludge。

由於整個HTTP請求都是一個事務，所以在發生衝突時它會回滾。我通過在整個HTTP請求的上下文中鎖定一個特殊的實體來使用它。如果同時收到多個HTTP請求，除1之外的所有請求都將導致一些PersistenceException。

在交易開始時，我檢查是否還沒有提交其他正確答案。在檢查期間，鎖已經有效，所以不會出現競賽狀況。在提交答案之前鎖一直有效。這基本上將關鍵部分模擬爲應用程序級別的SELECT + INSERT兩步查詢（在純MySQL中，我將使用INSERT IF NOT EXISTS構造）。

這種方法有一些缺點。每當兩名學生同時提交答案時，其中一人將被拋出異常。這對性能和帶寬有壞處，因爲收到HTTP STATUS 409的學生必須重新提交答案。

爲了彌補後者，我會自動重試在隨機選擇的時間間隔之間在服務器端提交幾次答案。請參閱根據HTTP請求控制器代碼如下：

@Controller 
@RequestMapping("/users") 
public class UserActionsController { 
    @Autowired 
    private SessionRegistry sessionRegistry; 

    @Autowired 
    @Qualifier("authenticationManager") 
    private AuthenticationManager authenticationManager;  
    @Resource(name = "userActionsManager") 
    private UserActionsManagerInterface userManager; 
    @Resource(name = "databaseManager") 
    private DB db; 

    . 
    . 
    . 

    @RequestMapping(value = "/{username}/{courseCode}/missions/{missionName}/tasks/{taskCode}/submitAnswer", method = RequestMethod.POST) 
    public @ResponseBody 
    Map<String, Object> giveAnswer(@PathVariable String username, 
      @PathVariable String courseCode, @PathVariable String missionName, 
      @PathVariable String taskCode, @RequestParam("answer") String answer, HttpServletRequest request) { 
     init(request); 
     db.log("Submitting an answer to task `"+taskCode+"` of mission `"+missionName+ 
       "` in course `"+courseCode+"` as student `"+username+"`."); 
     String str = null; 
     boolean conflict = true; 

     for (int i=0; i<10; i++) { 
      Random rand = new Random(); 
      int ms = rand.nextInt(1000); 

      try { 
       str = userManager.giveAnswer(username, courseCode, missionName, taskCode, answer); 
       conflict = false; 
       break; 
      } 
      catch (EntityExistsException e) {throw new EntityExistsException();} 
      catch (PersistenceException e) {} 
      catch (UnexpectedRollbackException e) {} 

      try { 
       Thread.sleep(ms); 
      } catch(InterruptedException ex) { 
       Thread.currentThread().interrupt(); 
      } 
     } 
     if (conflict) str = userManager.giveAnswer(username, courseCode, missionName, taskCode, answer); 

     if (str == null) db.log("Answer accepted: `"+answer+"`."); 
     else    db.log("Answer rejected: `"+answer+"`."); 

     Map<String, Object> hm = new HashMap<String, Object>(); 

     hm.put("success", str == null); 
     hm.put("message", str); 

     return hm; 
    } 
} 

如果由於某種原因，控制器無法提交事務10次在一排，然後它會嘗試一個更多的時間，但不會嘗試捕捉可能的例外。當第11次嘗試引發異常時，它將由全局異常控制器處理，客戶端將收到HTTP STATUS 409.全局異常控制器定義如下。

@ControllerAdvice 
public class GlobalExceptionController { 
    @Resource(name = "staticDatabaseManager") 
    private StaticDB db; 

    @ExceptionHandler(SessionAuthenticationException.class) 
    @ResponseStatus(value=HttpStatus.FORBIDDEN, reason="session has expired") //403 
    public ModelAndView expiredException(HttpServletRequest request, Exception e) { 
     ModelAndView mav = new ModelAndView("exception"); 
     mav.addObject("name", e.getClass().getSimpleName()); 
     mav.addObject("message", e.getMessage()); 
     return mav; 
    } 

    @ExceptionHandler({UnexpectedRollbackException.class, 
         EntityExistsException.class, 
         OptimisticLockException.class, 
         PersistenceException.class}) 
    @ResponseStatus(value=HttpStatus.CONFLICT, reason="conflicting requests") //409 
    public ModelAndView conflictException(HttpServletRequest request, Exception e) { 
     ModelAndView mav = new ModelAndView("exception"); 
     mav.addObject("name", e.getClass().getSimpleName()); 
     mav.addObject("message", e.getMessage()); 

     synchronized (db) { 
      db.setUserInfo(request); 
      db.log("Conflicting "+request.getMethod()+" request to "+request.getRequestURI()+" ("+e.getClass().getSimpleName()+").", Log.LVL_SECURITY); 
     }   

     return mav; 
    } 

    //ResponseEntity<String> customHandler(Exception ex) { 
    // return new ResponseEntity<String>("Conflicting requests, try again.", HttpStatus.CONFLICT); 
    //} 
} 

最後，giveAnswer方法本身利用具有主鍵lock_addCorrectAnswer特殊實體。我使用OPTIMISTIC_FORCE_INCREMENT標誌鎖定該特殊實體，確保沒有兩個事務對於giveAnswer方法可以有重疊的執行時間。相應的代碼可以看到下面：

@Component("userActionsManager") 
@Transactional 
public class UserActionsManager implements UserActionsManagerInterface { 

    @PersistenceContext(unitName = "itsadDB") 
    private EntityManager manager; 

    @Resource(name = "databaseManager") 
    private DB db; 

    . 
    . 
    . 

    @SuppressWarnings("unchecked") 
    @Override 
    @PreAuthorize("hasRole('ROLE_USER') && #username == authentication.name") 
    public String giveAnswer(String username, String courseCode, String missionName, String taskCode, String answer) { 
     . 
     . 
     . 
     if (!userCanGiveAnswer(user, course, missionTask)) { 
      error = "It is forbidden to submit an answer to this task."; 
      db.log(error, Log.LVL_MAJOR); 
      return error; 
     } 
     . 
     . 
     . 
     if (correctAnswer) { 
      . 
      . 
      .   
      addCorrectAnswer(newSubmission, result); 
      return null; 
     } 

     newSubmission = new Submission(user, course, missionTask, answer, false); 
     manager.persist(newSubmission); 
     return error; 
    } 

    private void addCorrectAnswer(Submission submission, Result result) { 
     String var = "lock_addCorrectAnswer"; 
     Global global = manager.find(Global.class, var); 

     if (global == null) { 
      global = new Global(var, 0); 
      manager.persist(global); 
      manager.flush(); 
     } 
     manager.lock(global, LockModeType.OPTIMISTIC_FORCE_INCREMENT); 
     manager.persist(submission); 
     manager.persist(result); 
     manager.flush(); 

     long submissions = getCorrectSubmissionCount(submission); 
     long results  = getResultCount(result); 
     if (submissions > 1 || results > 1) throw new EntityExistsException(); 
    } 

    private long getCorrectSubmissionCount(Submission newSubmission) { 
     Query q = manager.createQuery("SELECT count(s) FROM Submission AS s WHERE s.missionTask = ?1 AND s.course = ?2 AND s.user = ?3 AND s.correct = true"); 
     q.setParameter(1, newSubmission.getMissionTask()); 
     q.setParameter(2, newSubmission.getCourse()); 
     q.setParameter(3, newSubmission.getUser()); 
     return (Long) q.getSingleResult(); 
    } 

    private long getResultCount(Result result) { 
     Query q = manager.createQuery("SELECT count(r) FROM Result AS r WHERE r.missionTask = ?1 AND r.course = ?2 AND r.user = ?3"); 
     q.setParameter(1, result.getMissionTask()); 
     q.setParameter(2, result.getCourse()); 
     q.setParameter(3, result.getUser()); 
     return (Long) q.getSingleResult(); 
    } 
} 

值得注意的是，實體Global必須有國際級的版本註釋爲OPTIMISTIC_FORCE_INCREMENT工作是很重要的（見下面的代碼）。

@Entity 
@Table(name = "GLOBALS") 
public class Global implements Serializable { 
    . 
    . 
    . 
    @Id 
    @Column(name = "NAME", length = 32) 
    private String key; 
    @Column(name = "INTVAL") 
    private int intVal; 
    @Column(name = "STRVAL", length = 4096) 
    private String strVal; 
    @Version 
    private Long version; 
    . 
    . 
    . 
} 

這樣的方法可以進一步優化。對於所有的giveAnswer調用，我都可以使用相同的鎖名稱lock_addCorrectAnswer，而不必使用提交用戶的名稱確定性地生成鎖名。例如，如果學生的用戶名是Hyena那麼鎖實體的主鍵將是lock_Hyena_addCorrectAnswer。這樣，多名學生可以同時提交答案而不會發生任何衝突。但是，如果惡意用戶並行地使用HTTP POST方法對submitAnswer 10x進行垃圾郵件發送，則會被此鎖定機制阻止。