Spring/Spring Boot

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

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

Redis

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

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

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

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

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

考慮到這些實(shí)際場(chǎng)景，有時(shí)候需要異步執(zhí)行任務(wù)，減少 API 響應(yīng)時(shí)間。例如，如果一次在同一個(gè) API 調(diào)用中刪除一千多條記錄，那么肯定會(huì)增加 API 響應(yīng)時(shí)間。為了減少 API 響應(yīng)時(shí)間，可以運(yùn)行一個(gè)后臺(tái)任務(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 會(huì)在需要執(zhí)行特定操作時(shí)查找記錄，例如查找所有已發(fā)貨7天但未生成發(fā)票的記錄。這些調(diào)度機(jī)制中大多數(shù)都會(huì)遇到擴(kuò)展問(wèn)題，在數(shù)據(jù)庫(kù)中掃描查找相關(guān)行或者記錄。多數(shù)情況會(huì)引發(fā)全表掃描，性能非常差。設(shè)想一下正在運(yùn)行的應(yīng)用程序與批處理系統(tǒng)使用相同的數(shù)據(jù)庫(kù)。

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

Rqueue

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

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

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

消息傳遞

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

需要的工具：

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

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

需要添加以下依賴：

1.Spring Data Redis

2.Spring Web

3.Lombok 等。

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

使用Rqueue開(kāi)發(fā)庫(kù)實(shí)現(xiàn)按任意延遲時(shí)間執(zhí)行任務(wù)。Rqueue 是一個(gè)基于 Spring 的異步任務(wù)執(zhí)行器，可以按照任意延遲執(zhí)行任務(wù)，它基于 Spring 消息庫(kù)并由 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 功能。出于測(cè)試目的，這里還將啟用 WEB MVC。

application 文件更新如下：

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

使用 Rqueue 添加任務(wù)非常簡(jiǎn)單，只要對(duì)方法添加 RqueueListener 注解。RqueuListener 注解提供了一些字段，可根據(jù)使用場(chǎng)景設(shè)置。對(duì)于延遲任務(wù)，需要設(shè)置 delayedQueue="true" 并且必須提供 value；否則注解將被忽略。value 是隊(duì)列名稱。設(shè)置 deadLetterQueue 可以將任務(wù)推送到另一個(gè)隊(duì)列。否則，執(zhí)行失敗時(shí)任務(wù)會(huì)被丟棄。還可以使用 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 類分別存儲(chǔ)電子郵件和發(fā)票數(shù)據(jù)。簡(jiǎn)單起見(jiàn)，這些類只包含少數(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ù)使用場(chǎng)景，可以采用多種方式設(shè)置任務(wù)，例如，對(duì)于重試可以使用方法重試計(jì)數(shù)，對(duì)于延遲任務(wù)可以設(shè)置延遲。

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

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

這里會(huì)在30秒內(nèi)生成發(fā)票。為此，在發(fā)票隊(duì)列上提交一個(gè)延遲30000（毫秒）的任務(wù)。另外，這里會(huì)嘗試在后臺(tái)發(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)行該程序。程序成功啟動(dòng)后，訪問(wèn) 鏈接。

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

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

總結(jié)

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