定時任務使用spring-task，分布式是基于redis實現，可保證同一任務同時只有一個節點在執行。 ***** **使用技術：** * 分布式鎖，基于redis實現 * 隊列，基于redis實現 * 發布/訂閱，基于redis實現 * 分布式緩存，基于redis實現 ***** 實現原理： 1. 每個啟動的節點服務都執行一個內置任務，每秒執行一次，基于分布式鎖嘗試鎖定自己為master節點，集群環境中只有一個master節點。 2. master節點的每個定時任務執行時向隊列推入一個令牌，同時向集群所有節點廣播一條消息，各節點收到消息后從隊列中搶令牌，誰搶到誰執行真正的任務 ``` /** * 嘗試鎖定本服務作為master * * @param timeoutMillis */ @SuppressWarnings("unchecked") public void tryLockMaster(final long timeoutMillis) { final ICache healthCache = cacheManager.getICache(TaskConstants.TASK_HEALTH_CACHE_NAME); if (healthCache.getNativeCache() instanceof RedisOperations) { RedisOperations redisOperations = (RedisOperations) healthCache.getNativeCache(); //利用redis全局鎖 RedisLock lock = new RedisLock(redisOperations, TaskConstants.TASK_HEALTH_CACHE_NAME); lock.execute(new LockCallback<Boolean>() { @Override public Boolean doInLock(RedisConnection connection) { if (!existMaster() || isMaster()) { //將全局唯一ID置入緩存 healthCache.put(TaskConstants.TASK_HEALTH_KEY_NAME, TaskConstants.UU_ID); //設置過期時間 healthCache.expire(TaskConstants.TASK_HEALTH_KEY_NAME, timeoutMillis / 1000); log.info("I'm master! UUID[{}]", TaskConstants.UU_ID); } return true; } }); } } ``` ``` /** * 集群中本任務只有一個服務(master)在執行 * * @return */ @SuppressWarnings("unchecked") public void doTask() { MasterFactory masterFactory = getMasterFactory(); //如果使用集群模式則發送消息 if (masterFactory.isCluster()) { if (masterFactory.isMaster()) { //令牌入隊。使用隊列保證集群中同時只有一個服務執行任務，即集群中只有拿到令牌的服務才會執行任務 masterFactory.getQueueManager().getQueue(CURRENT_QUEUE).offer(1); //發送執行任務命令 masterFactory.getPubSubManager().publish(1, CURRENT_CHANNEL); } } //否則直接執行任務 else { doExecute(); } } ``` ``` /** * 接收消息并處理 * * @param channel * @param messageBody * @param pattern */ @SuppressWarnings("unchecked") @Override public void onMessage(String channel, Object messageBody, String pattern) { if (CURRENT_CHANNEL.equals(channel)) { //使用隊列保證集群中同時只有一個服務執行任務 IQueueManager queueManager = masterFactory.getQueueManager(); IQueue queue = queueManager.getQueue(CURRENT_QUEUE); if (queue.poll() != null) { //同步執行任務，即同一個任務直到上一次執行完成時再執行下次。單機運行時，spring task內部已經實現，但集群任務下需要另外實現，此處基于redis全局鎖實現 if (queueManager instanceof RedisQueueManager) { //基于redis全局鎖實現 RedisOperations redisOperations = ((RedisQueueManager) queueManager).getRedisOperations(); RedisLock lock = new RedisLock(redisOperations, CURRENT_TASK_LOCK_NAME); lock.execute(new LockCallback<Boolean>() { @Override public Boolean doInLock(RedisConnection connection) { //執行具體的任務 doExecute(); return true; } }); } //其他方式待實現... else { doExecute(); } } } } ``` 要想啟用分布式任務模式需設置以下參數 --cache.cacheDriver=rediscache --mq.pubsub.import=true --mq.queue.import=true --redis.host=xxx.xxx.xxx.xxx