java - 每秒将数据保存到数据库，解决数据库连接丢失或缓慢的问题

Question

我正在使用 spring data jpa 和 mysql 编写 java 控制台应用程序，我正在尝试解决以下情况:

应用程序每秒生成新对象，应在同一时刻以相同顺序保存到数据库。如果长时间保存对象期间数据库连接丢失或调用超时异常，则将即将到来的对象保存到临时缓冲区。当连接恢复时，将所有这些累积的对象和新即将生成的对象(在该特定时刻)保存到数据库。

我的问题是:

• 当数据库连接丢失时，如何处理将对象保存在临时缓冲区中？

我想当数据库连接丢失时我应该使用 ScheduledExecutorService 捕获 Throwable，然后将特定对象保存到 CopyOnWriteArrayList，这是正确的方法吗？

• 当连接丢失或先前的对象保存调用超时异常时，如何停止将新对象保存到数据库，并在连接建立时恢复保存即将到来的对象的过程？
• 当连接建立时，如何在保存新生成的对象之前将所有累积的对象保存到数据库？

更新

我编写了使用上述行为运行对象生成的服务:

服务

@Service
public class ReportService implements IReportService {

  @Autowired
  private ReportRepository reportRepository;

  @Override
  public void generateTimestamps() {

    BlockingQueue<Report> queue = new LinkedBlockingQueue<>();
    new Thread(new ReportsProducer(queue)).start();
    new Thread(new ReportsConsumer(queue, reportRepository)).start();
  }

  @Override
  public List<Report> showTimestamps() {
    return reportRepository.findAll();
  }
}

对象生成器:

public class ReportsProducer implements Runnable {

  private final BlockingQueue<Report> reportsQueue;

  ReportsProducer(BlockingQueue<Report> numbersQueue) {
    this.reportsQueue = numbersQueue;
  }

  public void run() {
    try {
      while (true) {
        generateReportEverySecond();
      }
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  private void generateReportEverySecond() throws InterruptedException {
    Thread.sleep(1000);
    Report report = new Report();
    reportsQueue.put(report);
    System.out.println(Thread.currentThread().getName() + ": Generated report[id='" + report.getId() + "', '" + report
            .getTimestamp() + "']");
  }
}

对象消费者:

public class ReportsConsumer implements Runnable {
  private final BlockingQueue<Report> queue;

  private ReportRepository reportRepository;

  ReportsConsumer(BlockingQueue<Report> queue, ReportRepository reportRepository) {
    this.queue = queue;

//    Not sure i do this correct way
    this.reportRepository = reportRepository;
  }

  public void run() {
    while (true) {
      try {
        if (!queue.isEmpty()) {
          System.out.println("Consumer queue size: " + queue.size());

          Report report = reportRepository.save(queue.peek());
          queue.poll();
          System.out.println(Thread.currentThread().getName() + ": Saved report[id='" + report.getId() + "', '" + report
                  .getTimestamp() + "']");
        }

      } catch (Exception e) {

        //  Mechanism to reconnect to DB every 5 seconds
        try {
          System.out.println("Retry connection to db");
          Thread.sleep(5000);
        } catch (InterruptedException e1) {
          e1.printStackTrace();
        }
      }
    }
  }
}

存储库:

@Repository
public interface ReportRepository extends JpaRepository<Report, Long> {
}

对象:

@Entity
@Table(name = "reports")
public class Report {

  @Id
  @GeneratedValue
  private Long id;

  @Column
  private Timestamp timestamp;

  public Report() {
    this.timestamp = new Timestamp(new Date().getTime());
  }

  public Long getId() {
    return id;
  }

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

  public Timestamp getTimestamp() {
    return timestamp;
  }

  public void setTimestamp(Timestamp timestamp) {
    this.timestamp = timestamp;
  }
}

应用程序属性:

spring.datasource.url=jdbc:mysql://xyz:3306/xyz
spring.datasource.username=xyz
spring.datasource.password=xyz

spring.jpa.properties.hibernate.dialect = org.hibernate.dialect.MySQL5Dialect
spring.jpa.properties.hibernate.connection.driver_class=com.mysql.cj.jdbc.Driver
spring.jpa.properties.hibernate.ddl-auto = create-drop

# Below properties don't work actually
spring.jpa.properties.javax.persistence.query.timeout=1
# Reconnect every 5 seconds
spring.datasource.tomcat.test-while-idle=true
spring.datasource.tomcat.time-between-eviction-runs-millis=5000
spring.datasource.tomcat.validation-query=SELECT 1

build.gradle:

version '1.0'

buildscript {
    ext {
        springBootVersion = '1.5.9.RELEASE'
    }
    repositories {
        mavenCentral()
    }
    dependencies {
        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
    }
}

apply plugin: 'java'
apply plugin: 'org.springframework.boot'

sourceCompatibility = 1.8
targetCompatibility = 1.8

ext {
    mysqlVersion = '6.0.6'
    dbcp2Version = '2.2.0'
    hibernateVersion = '5.2.12.Final'
}

repositories {
    mavenCentral()
}

dependencies {
    compile group: 'mysql', name: 'mysql-connector-java', version: mysqlVersion
    compile group: 'org.hibernate', name: 'hibernate-core', version: hibernateVersion
    compile group: 'org.hibernate', name: 'hibernate-c3p0', version: hibernateVersion
    compile("org.springframework.boot:spring-boot-starter-data-jpa")
    testCompile("org.springframework.boot:spring-boot-starter-test")
    compile group: 'org.assertj', name: 'assertj-core', version: '3.9.0'
}

根据上面的代码，我想知道一些时刻:

• 您建议如何检查数据库连接？

  我使用保存操作来检查，如果数据库连接丢失，我只需等待 5 秒并重复该操作。此外，数据源配置为在数据库连接关闭时每 5 秒建立一次。有更正确的方法吗？

• 如果数据库连接处于 Activity 状态，但数据库此时非常慢或太忙(ovberloaded)，会发生什么？

  据我了解，我需要设置查询超时。在这种情况下我还应该采取什么措施？

Answer 1

对于前三个问题，您正在寻找队列。如果您使用 Spring 框架，它确实为 JPA 使用单个类，因为 Repository 是一个 Bean，只要您配置它，Spring 就会为您处理数据库连接池正确。

1. 您建议如何检查数据库连接？

   您在 application.properties 中所做的操作确实检查了连接性。我认为 5000 毫秒太频繁，它可能会减慢您的系统速度。 360000 可能是一个不错的间隔。

2. 如果数据库连接处于 Activity 状态，但数据库此时非常慢或太忙(overloaded)，会发生什么？

   配置连接池时，您可以设置以下属性:

   removeAbandoned - 如果我们想要检测泄漏的连接，则设置为 true

   removeAbandonedTimeout - 从调用 dataSource.getConnection 到我们认为它被放弃的秒数

   logAbandoned - 如果我们应该记录连接被放弃，则设置为 true。如果此选项设置为 true，则会在 dataSource.getConnection 调用期间记录堆栈跟踪，并在未返回连接时打印堆栈跟踪

引用:Configuring jdbc-pool for high-concurrency