在区块链的世界里，常常很多时候用户需要充值，要不拉起钱包，要不支付到某个特定账号，这个时候可以监控合约交易记录实现实时到账，有的时候上某些网站的时候，至于是哪些网站，小编就不太好说了，有见过直接备注信息充值扫码支付到个人二维码，然后立马就会充值成功，那么这个是怎么实现的呢

当然是后台一个守护进程，然后实时监控到账情况，通过MEMO进行订单信息识别入款，springboot里新开一个守护进程很多种方式，但是小编我比较懒，就给大家介绍一种一个地方改代码就能实现的方式：

通过实现ApplicationRunner的多线程方式:

RechargeMonitorRunne.java

package io.xxxschedule;import io.xxx.component.RechargeMonitorServer;
import io.xxx.configuration.AsyncConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;@Component
public class RechargeMonitorRunner implements ApplicationRunner {private static final Logger logger = LoggerFactory.getLogger(RechargeMonitorRunner.class);@Resourceprivate RechargeMonitorServer monitorServer;@Resourceprivate AsyncConfiguration threadConfig;@Overridepublic void run(ApplicationArguments args) throws Exception {SimpleDateFormat df = new SimpleDateFormat("HH:mm");//设置日期格式logger.info("计时器开启。。。。。。");while (true){monitoServer.tradeMonitor();// this.tradeMonitor();String s=df.format(new Date());//System.out.println(s);if ("23:50".equals(s)){//当时间为23:50logger.info("当前时间为："+df.format(new Date())+"开启数据入库--------");break;//退出这个while(true)循环}}}@Asyncpublic Future<String> tradeMonitor() {System.out.println("\n\n----------------------------------------------");System.out.println(Thread.currentThread().getName() + "正在处理请求 tradeMonitor");System.out.println("----------------------------------------------");String result = "请求失败";//....你的业务逻辑//  return CompletableFuture.completedFuture(result);return null;}}

监控合约接口，当然也是一个内部api，异步执行，配置异步多线程池:

AsyncConfiguration.java

package io.xxx.configuration;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Executor;@Configuration
@EnableAsync  // 启用异步任务
public class AsyncConfiguration {// 声明一个线程池(并指定线程池的名字)@Bean("AsyncThread")public Executor asyncExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();//核心线程数5：线程池创建时候初始化的线程数executor.setCorePoolSize(5);//最大线程数5：线程池最大的线程数，只有在缓冲队列满了之后才会申请超过核心线程数的线程executor.setMaxPoolSize(10);//缓冲队列500：用来缓冲执行任务的队列executor.setQueueCapacity(500);//允许线程的空闲时间60秒：当超过了核心线程出之外的线程在空闲时间到达之后会被销毁executor.setKeepAliveSeconds(60);//线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池executor.setThreadNamePrefix("AsyncThread-");executor.initialize();return executor;}
}

业务实现monitorServer.java,这个异步执行文件需要另外一个类分开才能执行:

package io.xxx.component;import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;import javax.servlet.http.HttpSession;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;@Service
public class RechargeMonitorServer {@Async("AsyncThread")public void tradeMonitor() {System.out.println("\n\n----------------------------------------------");System.out.println(Thread.currentThread().getName() + "正在处理请求 tradeMonitor");System.out.println("----------------------------------------------");String result = "请求失败";//....你的业务逻辑//  return CompletableFuture.completedFuture(result);}
}

一run,是可以跑起来，哐当，隔一会儿就会报错，报错信息如下：

2020-11-26 09:50:58.061  INFO 18692 --- [           main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Shutting down ExecutorService 'AsyncThread'
RechargeMonitorRunner:Caused by: org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@6c4ce583[Running, pool size = 10, active threads = 10, queued tasks = 466, completed tasks = 271803]] did not accept task: org.springframework.aop.interceptor.AsyncExecutionInterceptor$$Lambda$654/1923626523@272778ae

原来while true里不停的加任务，任务满了就把这个守护线程挂了

怎么办，那么换一种写法捕捉队列满异常:

package io.xxx.schedule;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.xxx.Utils.ErrorAlarm;
import io.xxx.Utils.OkHttpUtils;
import io.xxx.component.RechargeMonitorServer;
import io.xxx.domain.Order;
import io.xxx.service.OrderService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.*;@Component
public class RechargeMonitorRunner implements ApplicationRunner {private static final Logger logger = LoggerFactory.getLogger(RechargeMonitorRunner.class);@Resourceprivate RechargeMonitorServer monitorServer;@Value("${xxx.rechargeMonitor.url}")private  String rechargeMonitorUrl;@Value("${xxx.rechargeMonitor.monitorAccount}")private  String eosMonitorAccount;private ConcurrentHashMap requestMap= new ConcurrentHashMap();private String requestJson = "";@Autowiredprivate OrderService orderService;@Autowiredprivate ErrorAlarm errorAlarm;@Overridepublic void run(ApplicationArguments args) throws Exception {SimpleDateFormat df = new SimpleDateFormat("HH:mm");//设置日期格式ThreadPoolTaskExecutor threadPoolTaskExecutor = initThreadPoolTaskExecutor();requestMap.put("code",eosMonitorAccount);requestMap.put("scope",eosMonitorAccount);requestMap.put("table","comein");requestMap.put("json",true);requestJson = JSON.toJSONString(requestMap);while (true) {//   System.out.println("Run start thread name->" + Thread.currentThread().getName());//List<Future> taskFutureList = new ArrayList<>();this.tradeMonitor();/*Future future = threadPoolTaskExecutor.submit(() -> {try {this.tradeMonitor();} catch (ExecutionException | InterruptedException e) {e.printStackTrace();}});taskFutureList.add(future);for (Future future2 : taskFutureList) {String result = (String) future2.get();System.out.println(result);}*/System.out.println("任务队列任务数量: " + threadPoolTaskExecutor.getThreadPoolExecutor().getQueue().size());//this.tradeMonitor();// monitorServer.tradeMonitor();String s = df.format(new Date());if ("23:50".equals(s)) {//当时间为23:50logger.info("当前时间为：" + df.format(new Date()) + "开启数据入库--------");break;//退出这个while(true)循环}// 获取任务队列中的任务数量}}@Asyncpublic Future<String> tradeMonitor() throws ExecutionException, InterruptedException {// System.out.println("Runner thread name->" + Thread.currentThread().getName());// System.out.println("\n\n----------------------------------------------");System.out.println(Thread.currentThread().getName() + "正在处理请求 tradeMonitor");System.out.println("----------------------------------------------");String result = "";result = OkHttpUtils.httpPostJson(rechargeMonitorUrl,requestJson);//result = null ;if(result==null  || result.equals("")){//报警ConcurrentHashMap alarmMap = new ConcurrentHashMap();alarmMap.put("api_url",rechargeMonitorUrl);alarmMap.put("params",requestJson);alarmMap.put("method","POST");alarmMap.put("message","xxx错误");int alarmFlag = errorAlarm.remoteRequestAlarm(alarmMap);result = "";}JSONObject jo = JSON.parseObject(result);CopyOnWriteArrayList<Order> orderIdList =  new CopyOnWriteArrayList<>();for(int i = 0;i<jo.getJSONArray("rows").size();i++){Order order = new Order();JSONObject oneRecord = (JSONObject)jo.getJSONArray("rows").get(i);String memo = oneRecord.getString("memo");String quantity = oneRecord.getString("quantity");if(memo.indexOf("orderID")>=0){String[] orderIdInfo = memo.split("\\:");Long orderID = Long.valueOf(orderIdInfo[1]);String[] amountInfo = quantity.split("\\s");BigDecimal amount = new BigDecimal(amountInfo[0]);if(amountInfo[1].equals("USDT") || amountInfo[1].equals("USDE")){order.setId(orderID);order.setHash(amount);}else {logger.info("RechargeMonitor record orderID:"+orderID+"is  NOTUSDTORUSDE");}orderIdList.add(order);}}int affectRows = orderService.confirmOrder(orderIdList);if(affectRows!=orderIdList.size()){logger.info("RechargeMonitor orderInfo :"+JSON.toJSONString(orderIdList)+"is  ORDERCONFIRMINCOMPLETE");}//  Thread.sleep(2000);// monitorServer.tradeMonitor();//....你的业务逻辑return CompletableFuture.completedFuture(result);// return null;}private ThreadPoolTaskExecutor initThreadPoolTaskExecutor() {ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();threadPoolTaskExecutor.setThreadNamePrefix("xxxTread-");threadPoolTaskExecutor.setCorePoolSize(5);threadPoolTaskExecutor.setMaxPoolSize(10);threadPoolTaskExecutor.setQueueCapacity(500);threadPoolTaskExecutor.setKeepAliveSeconds(60);threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy() {@Overridepublic void rejectedExecution(Runnable r, ThreadPoolExecutor e) {System.out.println("丢弃");logger.info("task full,reject,taskReject log");}});threadPoolTaskExecutor.initialize();return threadPoolTaskExecutor;}}

一跑，飞快，还能把CPU和内存尽可能的使用,当然如果不想即时监控，使用定时任务每秒跑一次也是可以的

最近天下不是很太平，开始招风了，也从来只相信共赢的信仰，任何歪门邪道拒不接受，大家可以关注我们的另外一个公众号“架构进阶”，扫码关注：

架构进阶

推荐阅读：Java8 Stream：2万字20个实例，玩转集合的筛选、归约、分组、聚合Java核心技术面试精讲【完结】Java并发和多线程基础面试题大集合【重磅分享】疯传25W次,BAT Java面试答案【283页PDF文档免费领】
小米用户画像实战，48页PPT下载
网易大数据用户画像实践《TensorFlow 2.0 深度学习算法实战》中文版教材电子版 pdf