SpringBoot集成XXL-JOB實現(xiàn)靈活控制的分片處理方案

更新時間：2024年09月26日 11:39:21 作者：碼到三十五

因為需要并行處理同一張數(shù)據(jù)表里的數(shù)據(jù),所以比較自然地想到了分片查詢數(shù)據(jù),可以利用對 id 取模的方法進行分片,避免同一條數(shù)據(jù)被重復處理,所以本文給大家介紹了SpringBoot集成XXL-JOB實現(xiàn)靈活控制的分片處理方案,需要的朋友可以參考下

場景

一個應用需要支持大量數(shù)據(jù)的批處理任務,要求：

并行處理能力：應用需能夠同時處理多個數(shù)據(jù)塊，即實現(xiàn)并行處理。
靈活的并發(fā)控制：可以靈活調(diào)整并行處理的任務數(shù)量，以確保資源利用最大化且不過載。
均衡負載分配：應將任務均勻分配到不同的服務器節(jié)點上，以平衡各節(jié)點的負載，避免單點壓力過大。

解決思路

因為需要并行處理同一張數(shù)據(jù)表里的數(shù)據(jù)，所以比較自然地想到了分片查詢數(shù)據(jù)，可以利用對 id 取模的方法進行分片，避免同一條數(shù)據(jù)被重復處理。那XXL-JOB 的路由策略「分片廣播 & 動態(tài)分片」很貼合這種場景」來調(diào)度定時任務；

實現(xiàn)DEMO

SpringBoot環(huán)境下，我們集成xxl-job來實現(xiàn)上述方案。
SpringBoot如何集成xxl-job查看官網(wǎng)即可，這里不再敘述，下面看下分片調(diào)度的代碼：

1.xxl-job調(diào)度管理頁面配置分片調(diào)度任務

路由策略選擇: 分片廣播

2. 編寫task代碼:

要獲取分片總數(shù)和當前分片序號,作為參數(shù)傳給sql語句:

	@Resource
    private OrderDataMapper orderDataMapper;
	
    @XxlJob("orderDataStatusTask")
    public void orderDataStatusTask() {
		// 計時器
        Stopwatch timer = Stopwatch.createStarted();
        
		// 獲取xxl-job的localThread中的總的分片數(shù)和當前分片
		OrderDataParam param = new OrderDataParam();
        param.setShardIndex(XxlJobHelper.getShardIndex());
        param.setShardTotal(XxlJobHelper.getShardTotal());
        // 其他參數(shù)設置,略了....
        
		// 根據(jù)分片數(shù)拉取當前分片的數(shù)據(jù)
        List<OrderData> orderDataList = orderDataMapper.getInitStatusOrder(param);
        XxlJobHelper.log("獲取待處理訂單數(shù)據(jù):分片號={},數(shù)據(jù)量={},總分片數(shù)={}", XxlJobHelper.getShardIndex(), orderDataList.size(), XxlJobHelper.getShardTotal());
        if (CollUtil.isEmpty(orderDataList)) {
            return;
        }
        // 處理邏輯,略了....
        
        XxlJobHelper.log("當前分片({})處理完成,耗時={}秒", XxlJobHelper.getShardIndex(), timer.stop().elapsed(TimeUnit.SECONDS));
    }

這里服務啟動了4個實例，總分片數(shù)ShardTotal就是4，每個實例的ShardIndex分別是0，1，2，3

3. mybatis中編寫sql語句

根據(jù)分片總數(shù)和當前分片數(shù)據(jù)對Id哈希取模, 這里做了兩次hash，主要作用是用id最后一位hash方便直接看出數(shù)據(jù)被哪個分片調(diào)度了。

	// 獲取未處理的訂單數(shù)據(jù)
	// 根據(jù)id末位數(shù)取hash后分片拉取
	<select id="getInitStatusOrder" parameterType="com.xxx.OrderDataParam"
            resultType="com.xxx.OrderData">
			select id,order_no,customer_code,
            from tt_order_data t
            where t.status = 0
				  and t.fail_count <![CDATA[ < ]]> #{retryCount}
				  and t.update_time <![CDATA[ >= ]]> #{lastUpdateTime}
				  and mod(mod(t.id, 10) , #{shardTotal}) = #{shardIndex}
			limit 0,200	
	</select>

4.最后看下調(diào)度日志

同一次調(diào)度任務，4個實例個調(diào)度一次，并且拉取到各自部分的數(shù)據(jù)進行處理：

第3個實例的調(diào)度日志：

	    2024-09-25 08:31:40 [com.xxl.job.core.thread.JobThread#run]-[130]-[Thread-144] 
		----------- xxl-job job execute start -----------
		----------- Param:{"lastHoursAgoModify":4,"rows":3000,"lastMonthAgoCreate":6,"retryCount":1}
		2024-09-25 08:31:40 [com.xxx.xxxx#orderDataStatusTask]-[47]-[Thread-144] 獲取待處理訂單數(shù)據(jù):分片號=3,數(shù)據(jù)量=100,總分片數(shù)=4
		2024-09-25 08:31:41 [com.xxx.xxxx#orderDataStatusTask]-[53]-[Thread-144] 當前分片(3)處理完成,耗時=1秒
		2024-09-25 08:31:41 [com.xxl.job.core.thread.JobThread#run]-[176]-[Thread-144] 
		----------- xxl-job job execute end(finish) -----------
		----------- Result: handleCode=200, handleMsg = null
		2024-09-25 08:31:41 [com.xxl.job.core.thread.TriggerCallbackThread#callbackLog]-[197]-[xxl-job, executor TriggerCallbackThread] 
        ----------- xxl-job job callback finish.

第4個實例的調(diào)度日志：

	    2024-09-25 08:31:40 [com.xxl.job.core.thread.JobThread#run]-[130]-[Thread-144] 
		----------- xxl-job job execute start -----------
		----------- Param:{"lastHoursAgoModify":4,"rows":3000,"lastMonthAgoCreate":6,"retryCount":1}
		2024-09-25 08:31:40 [com.xxx.xxxx#orderDataStatusTask]-[47]-[Thread-144] 獲取待處理訂單數(shù)據(jù):分片號=4,數(shù)據(jù)量=80,總分片數(shù)=4
		2024-09-25 08:31:41 [com.xxx.xxxx#orderDataStatusTask]-[53]-[Thread-144] 當前分片(4)處理完成,耗時=1秒
		2024-09-25 08:31:41 [com.xxl.job.core.thread.JobThread#run]-[176]-[Thread-144] 
		----------- xxl-job job execute end(finish) -----------
		----------- Result: handleCode=200, handleMsg = null
		2024-09-25 08:31:41 [com.xxl.job.core.thread.TriggerCallbackThread#callbackLog]-[197]-[xxl-job, executor TriggerCallbackThread] 
        ----------- xxl-job job callback finish.

到此這篇關于SpringBoot集成XXL-JOB實現(xiàn)靈活控制的分片處理方案的文章就介紹到這了,更多相關SpringBoot XXL-JOB分片處理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: