以管理倒计时为例：
package com.rongyi.platform.biotherm.web.websocket.activity.timers;
import com.rongyi.platform.biotherm.utils.StringUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.concurrent.*;
/**
* @desc: 周期性任务管理器。 是单实例的管理。 取消任务，需要通过MQ广播 或 调用者记录所在的实例
* @author: 毛会懂
* @create: 2019-10-13 10:35:00
**/
@Slf4j
@Service
public class ThreadExecutorManager {
// 周期任务线程池
private static ScheduledThreadPoolExecutor executorService = (ScheduledThreadPoolExecutor)Executors.newScheduledThreadPool(50);
// 管理定时任务的future
private static Map<String, Future> futureHashMap = new ConcurrentHashMap<>();
@PostConstruct
private void init(){
executorService.setRemoveOnCancelPolicy(true);
}
/**
* 添加定时器任务
*/
public static void addTimeTask(String timerName,Runnable runnable,Integer activityId,long period){
if(StringUtils.isEmpty(timerName) || activityId == null || runnable == null || period <= 0){
return ;
}
String realRankTaskName = realTaskName(timerName,activityId);
if(futureHashMap.get(realRankTaskName) != null){
// 在分布式系统中，并不能保证realRankTaskName 只有一个。需要调用者保证。
log.info("不应该执行到这里，同一个realRankTaskName只能有一个，添加失败");
return;
}
Future future = executorService.scheduleAtFixedRate(runnable, 0, period, TimeUnit.SECONDS);
futureHashMap.put(realRankTaskName, future);
log.info("添加定时器任务成功，taskName={}",realRankTaskName);
}
/**
* 删除定时器任务
*/
public static Boolean removeTimeTask(String timerName,Integer activityId){
if(StringUtils.isEmpty(timerName) || activityId == null){
return Boolean.TRUE;
}
String realRankTaskName = realTaskName(timerName,activityId);
Future removeFuture = futureHashMap.remove(realRankTaskName);
log.info("移除的定时任务名称：{}-{}",timerName,removeFuture);
if(removeFuture != null) {
boolean cancel = removeFuture.cancel(true);
log.info("任务取消是否成功：{}",cancel);
if(!cancel){
futureHashMap.put(realRankTaskName,removeFuture);
log.info("不应该执行到这里，任务取消失败，重新加入到futureHashMap中");
return Boolean.FALSE;
}
}
return Boolean.TRUE;
}
private static String realTaskName(String taskName,Integer activityId){
return taskName + "_" + activityId;
}
}
/**
* @desc: 向设备端推送倒计时
* @author: 毛会懂
* @create: 2019-10-13 09:46:00
**/
@Slf4j
public class CountDownExecutorService implements Runnable {
// 任务名
private String taskName;
// 活动id
private Integer activityId;
// 时长,递减到0，自动关闭
private Integer timeLength;
// screen不为空时，向大屏推送倒计时
private IDeviceScreen screen;
// mobile不为空时，向手机端推送倒计时
private IDeviceMobile mobile;
// 活动状态
private DeviceActivityStatusEnum activityStatus;
public CountDownExecutorService(String taskName, Integer activityId, Integer timeLength,
IDeviceScreen screen,
IDeviceMobile mobile,
IDeviceFoot foot,
DeviceActivityStatusEnum activityStatus){
this.taskName = taskName;
this.activityId = activityId;
this.timeLength = timeLength;
this.screen = screen;
this.mobile = mobile;
this.foot = foot;
this.activityStatus = activityStatus;
}
@Override
public void run() {
log.debug("倒计时开始,活动id：{},timeLength:{}",activityId,timeLength);
if(Thread.currentThread().isInterrupted()){
log.info("不应该执行到这里，已经检查到中断");
throw new RuntimeException();
}
try {
// 推送倒计时
pushCountDown();
//游戏结束,取消实时排行推送
if(timeLength <= 0){
ThreadExecutorManager.removeTimeTask(taskName,activityId);
return;
}
}catch (Exception e){
log.error("倒计时异常,活动id：{}，{}",activityId,e);
}
//计算剩余时间
timeLength--;
if(timeLength < 0){
log.info("不应该执行到这里，倒计时倒计到负数了");
throw new RuntimeException();
}
}
/**
* 向客户端推送倒计时。 可根据活动状态判断是否发送。但有可能引起性能问题。
*/
private void pushCountDown(){
if(screen != null){
screen.pushCountDown(activityId,timeLength,activityStatus);
}
if(mobile != null){
mobile.pushCountDown(activityId,timeLength,activityStatus);
}
}
}
调用方：
// 启动倒计时的定时器（用户已参与，准备中的倒计时）
Runnable realRankTask = new CountDownExecutorService(TASK_NAME_PRE,
activityId,
timeLength,
deviceScreen,
deviceMobile,
deviceFoot,
DeviceActivityStatusEnum.STATUS_PRE_MODE);
ThreadExecutorManager.addTimeTask(TASK_NAME_PRE,realRankTask,activityId,1);
// 取消定时任务：
webSocketSender.removeTimeTask(TASK_NAME_INIT_HIDDEN_QR,activityId);
/**
* 取消定时任务
*/
public Boolean removeTimeTask(String timerName,Integer activityId){
Map<String,Object> map = new HashMap<>();
map.put("timerName",timerName);
map.put("activityId",activityId);
rabbitTemplate.convertAndSend(RabbitMqConfig.PLATFORM_GAME_FANOUT_EXCHANGE_TASK,
RabbitMqConfig.PLATFORM_GAME_WEB_SOCKET_TIMER_QUEUE_KEY,
map);
log.info("取消定时任务的MQ发送成功：{}，{}",timerName,activityId);
return Boolean.TRUE;
}
取消定时任务MQ的接收：
@RabbitHandler
@RabbitListener(queues = "#{webSocketCancelTimerTask.name}")
public void consumerCancelTimerTask(Map<String,Object> map, Message message, Channel channel) {
MDC.put(Constant.TRACE_ID_NAME, UuidUtil.nextAsTextWithoutHyphen16());
final long deliveryTag = message.getMessageProperties().getDeliveryTag();
log.info("consumerCancelTimerTask，收到的信息：{}", JSON.toJSON(map));
try {
if(map == null){
channel.basicAck(deliveryTag, false);
return;
}
String timerName = (String) map.get("timerName");
Integer activityId = (Integer) map.get("activityId");
Boolean result = ThreadExecutorManager.removeTimeTask(timerName, activityId);
if (!result) {
throw new RuntimeException("移除倒计时定时器失败了");
}
// 通知 MQ 消息已被成功消费,可以ACK了
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
log.error("consumerCancelTimerTask 异常:", e);
try {
// 处理失败,重新压入MQ
channel.basicRecover();
Thread.sleep(100);
} catch (IOException | InterruptedException e1) {
log.error(e1.getMessage());
}
}
}
标签：map,定时器,timeLength,代码,完整,private,activityId,timerName,Integer From： https://www.cnblogs.com/maohuidong/p/17098583.html