Синхронизированный доступ к БД в Java

Question

У меня есть база данных, которая содержит отправленные электронные письма. Я использую несколько потоков для отправки этих сообщений электронной почты. Подход, который я использую, заключается в том, что каждый поток будет запрашивать базу данных, получать N сообщений электронной почты в памяти и маркировать их как отправленные. В другом потоке эти N-сообщения электронной почты будут отмечены как отмеченные и перейдут на следующую N записей.

Теперь это не работает, так как thread1 может обновлять записи как отправленные, запросы thread2 для электронной почты и, следовательно, оба потока в конечном итоге получают одинаковый набор сообщений электронной почты.

Каждый поток имеет свое собственное подключение к базе данных. Это основная причина такого поведения? Должен ли я просто делиться одним объектом соединения по всем потокам? Или есть лучший подход, который я мог бы использовать?

Answer 1

У вас должен быть способ поделиться одним методом // код для управления совпадением. Синхронизируйте заявления, чтобы получить электронные письма и пометить их. Затем отправили электронные письма. Что-то вроде этого:

public void processMails(){ 
    List<String> mails; 
    synchronized(this){ 
      mails = getMails(); 
      markMails(mails); 
     } 
     sendMails(mails); 

    } 

Этот метод может находиться на вашем DAO-фасаде, где доступны все потоки.

EDIT:

если у вас есть кратные экземпляры класса DAO:

public void processMails(){ 
     List<String> mails; 
     synchronize(DAO.class){ 
       mails = getMails(); 
       markMails(mails); 
      } 
      sendMails(mails); 

     } 

Другая альтернатива

private static final Object LOCK = new Object(); 

    public void processMails(){ 
      List<String> mails; 
      synchronize(LOCK){ 
        mails = getMails(); 
        markMails(mails); 
       } 
       sendMails(mails); 

      } 

Answer 2

Моя рекомендация состоит в том, чтобы иметь один поток заботиться о запросов к базе данных, размещение полученных писем в потокобезопасной очереди (например, ArrayBlockingQueue, которая имеет то преимущество, что она ограничена); вы можете получить любое количество потоков, удаляя и обрабатывая электронные письма из этой очереди. Накладные расходы на синхронизацию на ArrayBlockingQueue довольно легкие, и таким образом вам не нужно использовать транзакции базы данных или что-то в этом роде.

class EmailChunk { 
    Email[] emails; 
} 

// only instantiate one of these 
class DatabaseThread implements Runnable { 
    final BlockingQueue<EmailChunk> emailQueue; 

    public DatabaseThread(BlockingQueue<EmailChunk> emailQueue) { 
    this.emailQueue = emailQueue; 
    } 

    public void run() { 
    EmailChunk newChunk = // query database, create email chunk 

    // add newChunk to queue, wait 30 seconds if it's full 
    emailQueue.offer(newChunk, 30, TimeUnit.SECONDS); 
    } 
} 

// instantiate as many of these as makes sense 
class EmailThread implements Runnable { 
    final BlockingQueue<EmailChunk> emailQueue; 

    public EmailThread(BlockingQueue<EmailChunk> emailQueue) { 
    this.emailQueue = emailQueue; 
    } 

    public void run() { 
    // take next chunk from queue, wait 30 seconds if queue is empty 
    emailChunk nextChunk = emailQueue.poll(30, TimeUnit.SECONDS); 
    } 
} 

class Main { 
    final int queueSize = 5; 

    public static void main(String[] args) { 
    BlockingQueue<EmailChunk> emailQueue = new ArrayBlockingQueue<>(queueSize); 
    // instantiate DatabaseThread and EmailThread objects with this queue 
    } 
}