免费网页设计模板网站,免费站推广网站2022,wordpress自定义页面模板下载,想建一个网站怎么做文章目录前言一、异步队列实现思路#xff1f;二、实现步骤1.加入监听器2.实现监听器3.实现转交处理对象和转交处理线程4.自动转交异步处理5.业务如何添加前言
在某些场景下#xff0c;操作比较耗时#xff0c;给用户体验不是很好#xff0c;这时候我们就会直接想到两种方…
文章目录前言一、异步队列实现思路二、实现步骤1.加入监听器2.实现监听器3.实现转交处理对象和转交处理线程4.自动转交异步处理5.业务如何添加前言
在某些场景下操作比较耗时给用户体验不是很好这时候我们就会直接想到两种方案一种是定时任务一种就是异步队列那些实时性要求不高且比较耗时的任务是队列的最佳应用场景。 一、异步队列实现思路
持久化插入队列出队当程序突然停止当程序启动的时候从库里面拉出未执行的数据继续入队补偿机制下面是java的简单实现。
二、实现步骤
1.加入监听器 代码如下示例在web.xml 加入监听器 !--转交异步处理监听器 --listenerlistener-classcontext.ZjListener/listener-class/listener2.实现监听器
代码如下示例
public class ZjListener implements ServletContextListener {private static Logger log LoggerFactory.getLogger(ZjListener.class);Overridepublic void contextInitialized(ServletContextEvent servletContextEvent) {log.info(初始化转交异步处理线程...);ZjManager.getInstance().starup();log.info(初始化转交异步处理线程成功...);}Overridepublic void contextDestroyed(ServletContextEvent servletContextEvent) {ZjManager.getInstance().shutdown();}3.实现转交处理对象和转交处理线程
代码如下示例我就简单示例一些具体实现看业务昂
public class ZjRequset {private String dm;private String xh;public String getDm() {return dm;}public void setDm(String dm) {this.dm dm;}public String getXh() {return xh;}public void setXh(String xh) {this.xh xh;}
}public class ZjThread implements Runnable{private ZjRequset requset;private ZjService zjService WebAppContext.getBeanEx(ZjService);/*** description 实例化一个自动转交处理线程* param requset*/public ZjThread(ZjRequset requset){this.requset requset;}/*** description 获取请求* return ZjRequset*/public ZjRequset getRequest() {return requset;}Overridepublic void run() {if (requset ! null){zjService.saveZj(requset.getDm(), requset.getXh());}}
} 4.自动转交异步处理
代码如下示例这一块就是核心的代码了
public class ZjManager {private static final Log LOG LogFactory.getLog(ZjManager.class);private static final ZjManager INSTANCE new ZjManager();/*** 线程池维护线程的最少数量*/private final static int CORE_POOL_SIZE 2;/*** 线程池维护线程的最大数量*/private final static int MAX_POOL_SIZE 3;/*** 线程池维护线程所允许的空闲时间*/private final static int KEEP_ALIVE_TIME 0;/*** 线程池所使用的缓冲队列大小*/private final static int WORK_QUEUE_SIZE 200;/*** 是不是第一次启动程序*/private static boolean FIRST_QD true;/*** 自动转交异步处理队列*/private final QueueZjRequset requestQueue new LinkedListZjRequset();/*** 线程池*/private ThreadPoolExecutor threadPool null;/*** 调度器*/private ScheduledExecutorService scheduler null;/*** description 获取异步处理管理器实例*/public static ZjManager getInstance(){return INSTANCE;}/*** description 队列是否为空*/private boolean hasAcquire() {return !requestQueue.isEmpty();}/*** description 启动工作线程*/public boolean starup(){LOG.info( Console.getNowStr() 正在启动异步处理管理器...);threadPool new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,new ArrayBlockingQueueRunnable(WORK_QUEUE_SIZE), this.handler);scheduler Executors.newScheduledThreadPool(1);scheduler.scheduleAtFixedRate(accessBufferThread, 0, 3, TimeUnit.SECONDS);LOG.info(Console.getNowStr() 启动异步处理管理器成功!);return true;}/*** description 关闭工作线程*/public void shutdown(){if (scheduler ! null) {scheduler.shutdown();}if (threadPool ! null) {threadPool.shutdown();}}/*** description 处理器*/final RejectedExecutionHandler handler new RejectedExecutionHandler() {Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {synchronized (requestQueue) {try {requestQueue.offer(((ZjThread) r).getRequest());} catch (Exception e) {LOG.error(插入自动转交队列失败,e);}}}};/*** description 访问消息缓存的调度线程,查看是否有待定请求如果有则创建一个新的并添加到线程池*/final Runnable accessBufferThread new Runnable() {Overridepublic void run() {synchronized (requestQueue) {try {if (FIRST_QD){reloadRequest();}if (hasAcquire()) {ZjRequset request requestQueue.poll();if (request ! null) {Runnable task new ZjThread(request);threadPool.execute(task);}}} catch (Exception e) {LOG.error(重新执行失败,e);}}}};/*** 增加一个数据库操作** param request the request*/public void AddRequest(ZjRequset request) {try {if (request ! null) {//持久化写入库中wirteRequest(request);Runnable task new ZjThread(request);threadPool.execute(task);}} catch (Exception e) {LOG.error(e.getMessage(),e);}}/*** 写入到表中** param request the request*/void wirteRequest(ZjRequset request) {//将要请求写入库}/*** 将库中未执行的任务添加到队列中*/void reloadRequest() {FIRST_QD false;Connection conn null;PreparedStatement pst null;ResultSet rs null;//举个例子try {conn getConn();pst conn.prepareStatement(sql);rs pst.executeQuery();while (rs.next()) {ZjRequset request new ZjRequset();request.setDm(rs.getString(DM));request.setXh(rs.getString(XH));ZjManager.getInstance().AddRequest(request,);}} catch (SQLException e) {LOG.error(e.getMessage(), e);}finally {DBUtils.closeResultSet(rs);DBUtils.closePStatement(pst);DBUtils.closeConnection(conn);}}/*** 获取队列待处理线程数量*/public int getQueueCount(){return requestQueue.size();}/*** 获取处理线程的状态*/public int getThreadZt(){return threadPool.getActiveCount();}}5.业务如何添加 ZjRequset clRequest new ZjRequset();clRequest.setDm(dm);clRequest.setXh(xh);ZjManager.getInstance().AddRequest(clRequest);
