Spring/Spring Boot

Spring 是最流行 Java 應用程序開發(fā)框架。因此，Spring 社區(qū)也是最大的開源社區(qū)之一。除此之外，Spring 博客還提供了最新的開發(fā)文檔，內(nèi)容非常豐富。涵蓋了框架的內(nèi)部工作原理和示例項目，在StackOverflow上有10萬多個問題。

Spring 早期只支持基于XML的配置，為此飽受批評。后來 Spring 引入了基于注解的配置，情況發(fā)生了根本改變。Spring 3.0是第一個支持基于注解的配置的版本。2014年發(fā)布的 Spring Boot 1.0，徹底改變了人們對 Spring 框架生態(tài)的看法。在這里可以找到更詳細的時間表。

Redis

Redis 是最流行的 NoSQL 內(nèi)存數(shù)據(jù)庫之一，支持不同類型的數(shù)據(jù)結(jié)構(gòu)，包括 Set、哈希表、List、簡單鍵值對等。Redis 調(diào)用延遲為亞毫秒級，支持 replica set 等功能。Redis 操作的延遲也是亞毫秒級，在開發(fā)者社區(qū)中更具吸引力。

為什么需要異步執(zhí)行任務

一個典型的 API 調(diào)用包括以下五個方面：

1. 執(zhí)行數(shù)據(jù)庫（RDBMS 或 NoSQL）查詢
2. 在某些緩存系統(tǒng)（內(nèi)存、分布式等）上執(zhí)行操作
3. 進行計算（對一些數(shù)據(jù)進行數(shù)學計算）
4. 調(diào)用其他（內(nèi)部或外部）服務
5. 調(diào)度任務稍后執(zhí)行或者在后臺立即執(zhí)行。

任務可根據(jù)需要定時執(zhí)行，例如在創(chuàng)建訂單或裝運單后7天后生成發(fā)票。同樣，電子郵件通知無需立即發(fā)送，因此可以設(shè)為延期發(fā)送。

考慮到這些實際場景，有時候需要異步執(zhí)行任務，減少 API 響應時間。例如，如果一次在同一個 API 調(diào)用中刪除一千多條記錄，那么肯定會增加 API 響應時間。為了減少 API 響應時間，可以運行一個后臺任務刪除這些記錄。

延遲隊列

當計劃在指定時間或者按照設(shè)定間隔執(zhí)行任務時，可以使用 corn job。有很多不同的工具可以執(zhí)行定時任務，比如 UNIX 風格的 crontabs、Chronos。如果用 Spring 框架，那么可以用默認提供的 Scheduled 注解。

大多數(shù) cron job 會在需要執(zhí)行特定操作時查找記錄，例如查找所有已發(fā)貨7天但未生成發(fā)票的記錄。這些調(diào)度機制中大多數(shù)都會遇到擴展問題，在數(shù)據(jù)庫中掃描查找相關(guān)行或者記錄。多數(shù)情況會引發(fā)全表掃描，性能非常差。設(shè)想一下正在運行的應用程序與批處理系統(tǒng)使用相同的數(shù)據(jù)庫。

由于不可擴展，因此需要一些可擴展系統(tǒng)，可以在指定時間或按照設(shè)定時間間隔執(zhí)行任務，不會出現(xiàn)任何性能問題。有許多擴展的方法，例如用批處理方式執(zhí)行任務，或者在用戶、區(qū)域子集上執(zhí)行任務。另一種方法是在指定時間執(zhí)行任務，任務之間沒有依賴，例如 serverless 函數(shù)。定時器達到預定時間后會立即觸發(fā)執(zhí)行作業(yè)，這時可以使用延遲隊列。有很多隊列系統(tǒng)或軟件可供使用，但很少像 SQS 這樣可以設(shè)置延遲15分鐘，而不是延遲7個小時或者7天。

Rqueue

Rqueue 是針對 Spring 框架構(gòu)建的消息代理，它把數(shù)據(jù)存儲到 Redis 中并且提供了一種機制可以按任意延遲執(zhí)行任務。Rqueue 得到了 Redis 支持。相比 Kafka、SQS 等常見隊列系統(tǒng)，Redis 具有一些優(yōu)勢。在大多數(shù) Web 應用后端程序中，Redis 用來緩存數(shù)據(jù)或者其他用途。在當今世界，8.4％的 Web 應用程序正在使用 Redis 數(shù)據(jù)庫。

通常，使用 Kafka、SQS 或者其他隊列系統(tǒng)會帶來不同程度的額外開銷，而 Rqueue 和 Redis 可以將費用降為零。

除了使用 Kafka 帶來的開銷，還需配置基礎(chǔ)架構(gòu)、進行維護等等。由于大多數(shù)程序已經(jīng)使用了 Redis，因此不需要其他操作開銷。實際上，可以在同一個 Redis 服務器或群集上使用 Rqueue。Rqueue支持任意長度延遲

消息傳遞

由于長數(shù)據(jù)不會在數(shù)據(jù)庫中丟失，Rqueue 能夠確保至少發(fā)送一次消息。在 Rqueue 簡介中可以了解更多信息。

需要的工具：

1. IDE
2. Gradle
3. Java
4. Redis

簡單起見，這里使用 Spring Boot。通過Spring Boot初始化程序 創(chuàng)建一個 Gradle 項目。

需要添加以下依賴：

1.Spring Data Redis

2.Spring Web

3.Lombok 等。

目錄/文件夾結(jié)構(gòu)如下所示：

使用Rqueue開發(fā)庫實現(xiàn)按任意延遲時間執(zhí)行任務。Rqueue 是一個基于 Spring 的異步任務執(zhí)行器，可以按照任意延遲執(zhí)行任務，它基于 Spring 消息庫并由 Redis 提供支持。

這里將使用**com.github.sonus21:rqueue-spring-boot-starter:1.2-RELEASE添加 Rqueue spring boot starter **依賴。

dependencies {
  implementation 'org.springframework.boot:spring-boot-starter-data-redis'
  implementation 'org.springframework.boot:spring-boot-starter-web'
  implementation 'com.github.sonus21:rqueue-spring-boot-starter:1.2-RELEASE'
  compileOnly 'org.projectlombok:lombok'   
  annotationProcessor 'org.projectlombok:lombok'
  providedRuntime 'org.springframework.boot:spring-boot-starter-tomcat'
  testImplementation('org.springframework.boot:spring-boot-starter-test') {
    exclude group: 'org.junit.vintage', module: 'junit-vintage-engine'  
  }
}

需要啟用 Redis Spring Boot 功能。出于測試目的，這里還將啟用 WEB MVC。

application 文件更新如下：

@SpringBootApplication
@EnableRedisRepositories
@EnableWebMvc
public class AsynchronousTaskExecutorApplication {
  public static void main(String[] args) {
    SpringApplication.run(AsynchronousTaskExecutorApplication.class, args);
  }
}

使用 Rqueue 添加任務非常簡單，只要對方法添加 RqueueListener 注解。RqueuListener 注解提供了一些字段，可根據(jù)使用場景設(shè)置。對于延遲任務，需要設(shè)置 delayedQueue="true" 并且必須提供 value；否則注解將被忽略。value 是隊列名稱。設(shè)置 deadLetterQueue 可以將任務推送到另一個隊列。否則，執(zhí)行失敗時任務會被丟棄。還可以使用 numRetries 字段設(shè)置任務重試次數(shù)。

創(chuàng)建名為 MessageListener 的 Java 文件并增加一些方法執(zhí)行任務：

@Component
@Slf4j
public class MessageListener {
  @RqueueListener(value = "${email.queue.name}") (1)
  public void sendEmail(Email email) {
    log.info("Email {}", email);
  }
  @RqueueListener(delayedQueue = "true", value = "${invoice.queue.name}") (2)
  public void generateInvoice(Invoice invoice) {
    log.info("Invoice {}", invoice);
  }
}

用 Email 和 Invoice 類分別存儲電子郵件和發(fā)票數(shù)據(jù)。簡單起見，這些類只包含少數(shù)幾個字段。

Invoice.java：

import lombok.Data;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Invoice {
  private String id;
  private String type;
}

Email.java:

import lombok.Data;
@Data
@AllArgsConstructor
@NoArgsConstructor
public class Email {
  private String email;
  private String subject;
  private String content;
}

提交任務

可以使用 RqueueMessageSender bean 提交任務。根據(jù)使用場景，可以采用多種方式設(shè)置任務，例如，對于重試可以使用方法重試計數(shù)，對于延遲任務可以設(shè)置延遲。

需要對 RqueueMessageSender 進行 autowire 或使用構(gòu)造函數(shù)注入 bean。

下面展示了如何為測試創(chuàng)建 Controller。

這里會在30秒內(nèi)生成發(fā)票。為此，在發(fā)票隊列上提交一個延遲30000（毫秒）的任務。另外，這里會嘗試在后臺發(fā)送一封電子郵件。為此添加兩個 GET 方法，sendEmail 和 generateInvoice，當然也可以使用 POST。

@RestController
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
@Slf4j
public class Controller {
  private @NonNull RqueueMessageSender rqueueMessageSender;
  @Value("${email.queue.name}")
  private String emailQueueName;
  @Value("${invoice.queue.name}")
  private String invoiceQueueName;
  @Value("${invoice.queue.delay}")
  private Long invoiceDelay;
  @GetMapping("email")
  public String sendEmail(
      @RequestParam String email, @RequestParam String subject, @RequestParam String content) {
    log.info("Sending email");
    rqueueMessageSender.put(emailQueueName, new Email(email, subject, content));
    return "Please check your inbox!";
  }
  @GetMapping("invoice")
  public String generateInvoice(@RequestParam String id, @RequestParam String type) {
    log.info("Generate invoice");
    rqueueMessageSender.put(invoiceQueueName, new Invoice(id, type), invoiceDelay);
    return "Invoice would be generated in " + invoiceDelay + " milliseconds";
  }
}

在 application.properties 加入以下內(nèi)容：

email.queue.name=email-queue
invoice.queue.name=invoice-queue
# 30 seconds delay for invoice
invoice.queue.delay=300000

現(xiàn)在可以運行該程序。程序成功啟動后，訪問 鏈接。

在日志中，可以看到電子郵件任務正在后臺執(zhí)行：

下面是延遲30秒生成發(fā)票：

總結(jié)

可以看到，使用 Rqueue 執(zhí)行定時任務不需要冗長的模板代碼！配置和使用 Rqueue 庫時，進行了仔細考慮。要記?。簾o論是普通任務還是延遲任務，都需要盡快執(zhí)行。