java 使用Queue在队列中异步执行任务

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

先创建一个总的Handler（队列统一处理接口），名字就叫做 QueueTaskHandler

public interface QueueTaskHandler { void processData();
}

然后写一个队列服务类，就不多做说明了，我的注释已经写的很清楚了

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;@Component
public class QueueGenerationService{ // 日志监控
private static final Logger log = LoggerFactory.getLogger(QueueGenerationService.class);
// 根据业务与服务器性能自行配置这里我配置的是最多50000个任务
// LinkedBlockingQueue构造的时候若没有指定大小，则默认大小为Integer.MAX_VALUE
private final LinkedBlockingQueue<QueueTaskHandler> tasks = new LinkedBlockingQueue<QueueTaskHandler>(50000);
// 类似于一个线程总管保证所有的任务都在队列之中
private ExecutorService service = Executors.newSingleThreadExecutor();
// 检查服务是否运行
private volatile boolean running = true;
//线程状态
private Future<?> serviceThreadStatus = null; @PostConstruct
public void init() {
serviceThreadStatus = service.submit(new Thread(new Runnable() {
@Override
public void run() {
while (running) {
try {
//开始一个任务
QueueTaskHandler task = tasks.take();
try {
task.processData();
} catch (Exception e) {
log.error("任务处理发生错误", e);
}
} catch (InterruptedException e) {
log.error("服务停止，退出", e);
running = false;
}
}
}
}, "save data thread"));
} public boolean addData(QueueTaskHandler dataHandler) {
if (!running) {
log.warn("service is stop");
return false;
}
//offer 队列已经满了，无法再加入的情况下
boolean success = tasks.offer(dataHandler);
if (!success) {
log.warn("添加任务到队列失败");
}
return success;
}

//判断队列是否有任务
    public boolean isEmpty() {
   return tasks.isEmpty();
    }
public boolean checkServiceRun() {
return running && !service.isShutdown() && !serviceThreadStatus.isDone();
} public void activeService() {
running = true;
if (service.isShutdown()) {
service = Executors.newSingleThreadExecutor();
init();
log.info("线程池关闭，重新初始化线程池及任务");
}
if (serviceThreadStatus.isDone()) {
init();
log.info("线程池任务结束，重新初始化任务");
}
} @PreDestroy
public void destory() {
running = false;
service.shutdownNow();
}
}

接下来就可以开始写你的业务Handler了

public class TestServiceHandler implements QueueTaskHandler { // ******* start 这一段并不是必要的，这是示范一个传值的方式
private String name; private Integer age; public TestServiceHandler(String name) {
this.name = name;
} public TestServiceHandler(Integer age) {
this.age = age;
} public TestServiceHandler(String name, Integer age) {
this.name = name;
this.age = age;
} // ****** end // 这里也就是我们实现QueueTaskHandler的处理接口
@Override
public void processData() {
// 可以去做你想做的业务了
// 这里需要引用spring的service的话，我写了一个工具类，下面会贴出来
// ItestService testService = SpringUtils.getBean(ItestService.class);
System.out.println("name > " + name + "," + "age > " + age);
}}

那么我们来在service中添加一个任务

// 这里注入队列服务
@Autowired
private QueueGenerationService queueGenerationService;

// 在方法中调用与传参的方式
queueGenerationService.addData(new TestServiceHandler("小明",5));

整个过程就结束了，然后在你的业务Handler中如果需要使用其他的bean比如service，那么请试试我写的这个工具类

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;@Component
public class SpringUtils implements ApplicationContextAware { private static ApplicationContext applicationContext; /**
* @return
* @Description 获取applicationContext
*/
public static ApplicationContext getApplicationContext() {
return applicationContext;
} @Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
if (SpringUtils.applicationContext == null) {
SpringUtils.applicationContext = applicationContext;
}
} /**
* @param name
* @return
* @Description 通过name获取 Bean.
*/
public static Object getBean(String name) {
return getApplicationContext().getBean(name);
} /**
* @param clazz
* @return
* @Description 通过class获取Bean.
*/
public static <T> T getBean(Class<T> clazz) {
return getApplicationContext().getBean(clazz);
} /**
* @param name
* @param clazz
* @return
* @Description 通过name, 以及Clazz返回指定的Bean
*/
public static <T> T getBean(String name, Class<T> clazz) {
return getApplicationContext().getBean(name, clazz);
}}