SpringBoot與ElasticJob整合，實現一百萬條數據的狀態秒級更新

作者：Java知識日歷 2025-03-27 09:05:28

開發前端

隨著訂單數量急劇增加，傳統的單線程訂單狀態更新方式已經無法滿足高效處理的需求。為了提高訂單狀態更新的效率和系統的響應能力，我們決定采用分布式任務調度實現高效的訂單狀態批量更新。

應用場景

緩存預熱

緩存填充：在系統啟動時或高峰期來臨前，預先加載數據到緩存中，每個分片處理一部分數據。

用戶通知系統

郵件發送：向大量用戶發送電子郵件，可以通過分片來并行處理郵件發送任務，減少延遲。
推送通知：向移動設備推送通知，每個分片處理一部分用戶的推送任務。

消息隊列處理

消息處理：消費消息隊列中的消息，通過分片來提高消息處理的速度和效率。
事件驅動任務：根據事件觸發的任務，可以使用分片來并行處理不同的事件類型或事件來源。

數據批處理

批量數據導入/導出：將大量數據從一個系統遷移到另一個系統時，可以將數據分成多個批次進行處理。
日志處理：分析和處理大量的日志文件，每個分片處理一部分日志。

定時任務調度

定時數據同步：定期將數據從源數據庫同步到目標數據庫，可以通過分片來并行處理不同部分的數據。
報告生成：生成復雜的報表或統計信息，每個分片負責計算一部分數據的結果。

大規模數據分析

數據清洗：對海量數據進行清洗和預處理，每個分片處理一部分數據。
特征提取：從大數據集中提取特征，每個分片處理一部分數據集。

分布式爬蟲

網頁抓取：分布式爬蟲可以從多個節點同時抓取網頁內容，每個分片負責抓取一組URL。
數據采集：從不同的數據源收集數據，每個分片處理一個數據源。

游戲服務器管理

玩家數據更新：在游戲中，頻繁地更新玩家數據，可以通過分片來并行處理不同玩家的數據。
游戲邏輯計算：在多人在線游戲中，并行計算不同區域的游戲邏輯。

內容推薦系統

個性化推薦：為用戶提供個性化的推薦內容，每個分片處理一部分用戶的推薦任務。

實時監控與報警

監控指標收集：實時收集和處理監控指標，每個分片負責收集一部分系統的監控數據。
報警規則評估：評估報警規則，每個分片處理一部分報警條件。

任務分片的目的

負載均衡：通過將任務分配到多個節點上，避免單個節點過載。
提高性能：利用多核CPU或多臺機器的計算能力來加速任務執行。
容錯性：即使某個節點失敗，其他節點仍然可以繼續處理剩余的任務。

我們為什么選擇ElasticJob？

簡化任務開發

ElasticJob 提供了簡潔的任務接口，開發者只需關注具體的業務邏輯，而不必過多關心任務調度的底層細節。這大大提高了開發效率，減少了潛在的錯誤。

細粒度的日志記錄

ElasticJob 支持詳細的日志記錄，可以幫助開發者追蹤任務的執行過程，定位和解決問題。這對于調試和優化性能至關重要。

可靠的作業執行機制

ElasticJob具備強大的容錯能力和故障恢復機制。如果某個節點發生故障，其他節點可以接管其分片任務，確保任務的連續性和可靠性。這種設計使得系統能夠在面對突發情況時保持穩定運行。

動態分片策略

ElasticJob支持多種分片策略，可以根據實際需求調整分片的數量和分布方式。這對于處理不同規模的數據集非常有用。例如，在訂單數量增加時，可以通過簡單的配置調整分片總數，而無需對代碼進行大量修改。

可視化監控與管理

ElasticJob 提供了一個內置的控制臺，用于監控和管理任務的執行情況。通過這個控制臺，管理員可以實時查看任務的狀態、執行歷史、分片信息等，方便進行運維和調優。

支持定時任務

ElasticJob 內置了Cron表達式的支持，可以輕松地設置任務的執行時間表。這對于需要定期執行的任務（如每天或每小時執行一次的訂單狀態更新）非常方便。

如何進行任務分片？

確定分片總數：根據任務的特點和系統的資源情況決定需要分成多少個小任務。
分配分片項：為每個小任務分配一個唯一的分片項。
實現任務邏輯：編寫代碼來處理特定分片項對應的任務數據。
調度器管理：使用調度框架ElasticJob來管理和調度這些分片任務。

代碼實操

<!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- Spring Data JPA -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>

    <!-- MySQL Connector -->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <scope>runtime</scope>
    </dependency>

    <!-- ElasticJob Lite Core -->
    <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-core</artifactId>
        <version>2.1.5</version>
    </dependency>

    <!-- ElasticJob Lite Spring Boot Starter -->
    <dependency>
        <groupId>com.dangdang</groupId>
        <artifactId>elastic-job-lite-spring-boot-starter</artifactId>
        <version>2.1.5</version>
    </dependency>

    <!-- Zookeeper Client -->
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>4.2.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>4.2.0</version>
    </dependency>

application.yml

server:
  port:8080

spring:
datasource:
    url:jdbc:mysql://localhost:3306/order_db?useSSL=false&serverTimezone=UTC
    username:root
    password:123456
    driver-class-name:com.mysql.cj.jdbc.Driver

jpa:
    hibernate:
      ddl-auto:update
    show-sql:true
    properties:
      hibernate:
        dialect:org.hibernate.dialect.MySQL5InnoDBDialect

elasticjob:
regCenter:
    serverLists:localhost:2181# ZooKeeper服務器地址
    namespace:elastic-job-demo   # 命名空間
jobs:
    orderStatusUpdateJob:
      cron: 0 0 * * * ?     # Cron表達式，每小時執行一次
      shardingTotalCount:5   # 分片總數
      jobClass:com.example.job.OrderStatusUpdateJob# 任務類全限定名
      description:"更新訂單狀態"# 任務描述

Application.java

package com.example;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling  // 啟用定時任務調度
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

訂單狀態更新任務

package com.example.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.simple.api.SimpleJob;
import com.example.service.OrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 訂單狀態更新任務類
 * 實現SimpleJob接口，用于定期更新訂單狀態
 */
@Component
publicclass OrderStatusUpdateJob implements SimpleJob {

    privatestaticfinal Logger log = LoggerFactory.getLogger(OrderStatusUpdateJob.class);

    @Autowired
    private OrderService orderService;  // 注入OrderService服務

    /**
     * 執行任務的方法
     * @param context 分片上下文
     */
    @Override
    public void execute(ShardingContext context) {
        int shardingItem = context.getShardingItem();  // 獲取當前分片項
        int shardingTotalCount = context.getShardingTotalCount();  // 獲取總分片數
        long maxOrderId = orderService.getMaxOrderId();  // 獲取最大訂單ID

        // 計算當前分片需要處理的訂單范圍
        long startId = (long) shardingItem * (maxOrderId / shardingTotalCount);
        long endId = Math.min(startId + (maxOrderId / shardingTotalCount), maxOrderId);

        log.info("Processing orders from {} to {}", startId, endId);  // 記錄處理的訂單范圍
        int updatedCount = orderService.updateStatusInRange(startId, endId, "Processed");  // 更新訂單狀態

        log.info("Updated {} orders in range {} to {}", updatedCount, startId, endId);  // 記錄更新結果
    }
}

訂單實體類

package com.example.model;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

/**
 * 訂單實體類
 * 映射到數據庫中的order表
 */
@Entity
publicclass Order {

    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;  // 訂單ID

    private String status;  // 訂單狀態

    public Long getId() {
        return id;
    }

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

    public String getStatus() {
        return status;
    }

    public void setStatus(String status) {
        this.status = status;
    }
}

訂單Repository

package com.example.repository;

import com.example.model.Order;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Modifying;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

/**
 * 訂單Repository接口
 * 提供基本的CRUD操作和自定義查詢
 */
publicinterface OrderRepository extends JpaRepository<Order, Long> {

    /**
     * 根據ID范圍查找訂單
     * @param startId 開始ID
     * @param endId 結束ID
     * @return 訂單列表
     */
    List<Order> findByIdBetween(Long startId, Long endId);

    /**
     * 根據ID范圍更新訂單狀態
     * @param startId 開始ID
     * @param endId 結束ID
     * @param newStatus 新狀態
     * @return 更新的訂單數量
     */
    @Modifying
    @Transactional
    @Query("UPDATE Order o SET o.status = :newStatus WHERE o.id BETWEEN :startId AND :endId")
    int updateStatusInRange(@Param("startId") Long startId, @Param("endId") Long endId, @Param("newStatus") String newStatus);
}

訂單服務類

package com.example.service;

import com.example.model.Order;
import com.example.repository.OrderRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * 訂單服務類
 * 處理訂單相關的業務邏輯
 */
@Service
publicclass OrderService {

    @Autowired
    private OrderRepository orderRepository;  // 注入OrderRepository

    /**
     * 根據ID范圍獲取訂單
     * @param startId 開始ID
     * @param endId 結束ID
     * @return 訂單列表
     */
    public List<Order> getOrdersByRange(Long startId, Long endId) {
        return orderRepository.findByIdBetween(startId, endId);
    }

    /**
     * 根據ID范圍更新訂單狀態
     * @param startId 開始ID
     * @param endId 結束ID
     * @param newStatus 新狀態
     * @return 更新的訂單數量
     */
    public int updateStatusInRange(Long startId, Long endId, String newStatus) {
        return orderRepository.updateStatusInRange(startId, endId, newStatus);
    }

    /**
     * 獲取最大訂單ID
     * @return 最大訂單ID
     */
    public long getMaxOrderId() {
        return orderRepository.findAll().stream()
                .mapToLong(Order::getId)
                .max()
                .orElse(0L);
    }
}

責任編輯：武曉燕來源： Java知識日歷