Spring/Spring Boot

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

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

Redis

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

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

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

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

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

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

延遲隊(duì)列

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

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

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

Rqueue

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

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

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

消息傳遞

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

需要的工具：

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

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

需要添加以下依賴：

1.Spring Data Redis

2.Spring Web

3.Lombok 等。

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

使用Rqueue開發(fā)庫實(shí)現(xiàn)按任意延遲時(shí)間執(zhí)行任務(wù)。Rqueue 是一個(gè)基于 Spring 的異步任務(wù)執(zhí)行器，可以按照任意延遲執(zhí)行任務(wù)，它基于 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 添加任務(wù)非常簡單，只要對方法添加 RqueueListener 注解。RqueuListener 注解提供了一些字段，可根據(jù)使用場景設(shè)置。對于延遲任務(wù)，需要設(shè)置 delayedQueue="true" 并且必須提供 value；否則注解將被忽略。value 是隊(duì)列名稱。設(shè)置 deadLetterQueue 可以將任務(wù)推送到另一個(gè)隊(duì)列。否則，執(zhí)行失敗時(shí)任務(wù)會被丟棄。還可以使用 numRetries 字段設(shè)置任務(wù)重試次數(shù)。

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

@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ù)幾個(gè)字段。

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;
}

提交任務(wù)

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

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

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

這里會在30秒內(nèi)生成發(fā)票。為此，在發(fā)票隊(duì)列上提交一個(gè)延遲30000（毫秒）的任務(wù)。另外，這里會嘗試在后臺發(fā)送一封電子郵件。為此添加兩個(gè) GET 方法，sendEmail 和 generateInvoice，當(dāng)然也可以使用 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)在可以運(yùn)行該程序。程序成功啟動后，訪問 鏈接。

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

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

總結(jié)

可以看到，使用 Rqueue 執(zhí)行定時(shí)任務(wù)不需要冗長的模板代碼！配置和使用 Rqueue 庫時(shí)，進(jìn)行了仔細(xì)考慮。要記住：無論是普通任務(wù)還是延遲任務(wù)，都需要盡快執(zhí)行。