多线程异步调度任务

业务场景：Controller接受一个参数Object,调到业务层，把Object加入队列，成功就返回，同时开启线程做异步的消息发送。

目录结构：

CreatThreadPool.java

public class CreatThreadPool {

    private ThreadPoolExecutor pool;

    public CreateThreadPool(long maxAlive, int maxSize, int minSize, int queueSize) {
        pool = new ThreadPoolExecutor(minSize, maxSize, maxAlive, TimeUnit.SECONDS, new LinkedBlockingQueue(queueSize));
    }

    public Future<String> submit(Runnable task) {
        return pool.submit(task, null);
    }

    @ManagedAttribute
    public String getPoolStatus() {
        JSONObject json = new JSONObject();
        json.put("activeCount", pool.getActiveCount());
        json.put("taskCount", pool.getTaskCount());
        json.put("completedTaskCount", pool.getCompletedTaskCount());
        return json.toJSONString();
    }
}

CreatThread.java

public class CreatThread implements Runnable{

    private Object obj;
    
    private ApplicationContext context;

    @Override
    public void run() {

        try {
            sendMessage();
        } catch (Exception e) {
            e.getMessage();
        }
    }

    public void sendMessage() {
          //因为此处是线程类所以spring的bean加载不进来，需要使用ApplicationContext来获取
        MQPublisher mQPublisher=(MQPublisher) context.getBean(MQPublisher.class);

        Object obj=(MQLinkDTO) context.getBean(Object.class);
        //此处是为了把对象转换为JSON格式，所以用了阿里的fastjson
        JSONObject json=(JSONObject) JSON.toJSON(noticeDTO);
        
        mQPublisher.sendMessage(mQLinkDTO.getName(), mQLinkDTO.getKey(),json);
    }

    public void setObject(Object obj) {
        this.obj = obj;
    }
    
    public void setContext(ApplicationContext context) {
        this.context = context;
    }
}

Disptch.java

public class Dispatch implements Runnable,ApplicationContextAware{

    
    private LinkedBlockingQueue<Object> queue;
    
    private ApplicationContext context;
    //注入一个空的线程池
    @Autowired
    private CreatThreadPool pool;
    
    public Dispatcher(int queueSize) {
        //初始化队列
        queue=new LinkedBlockingQueue<>(queueSize);
    }

    @Override
    public void run() {
        while (true) {
            Object obj = null;
            try {
                obj = queue.take();
                
                if (obj != null) {
               //创建线程，把object传过去，并且把context传过去，把线程放到线程池，就绪
                    CreatThread thread = new CreatThread();
                    thread.setObject(obj);
                    thread.setContext(context);
                    pool.submit(thread);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        
    }

    public boolean offer(Object obj) {
        //将obj加入队列，成功返回true
        boolean result = queue.offer(obj);
        return result;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
       //获取上下文对象
        this.context=applicationContext;
    }
}

LoadedConfiguration.java

@Configuration
public class LoadedConfiguration {

//读取配置文件，生产Pool、Dispatcher、MQLinkDTO的bean
    @Bean(name="CreatThreadPool")
    public CreatThreadPool loadThreadPool() {
        return new CreatThreadPool(
                Long.parseLong((String) System.getProperty("threadpool.maxAliveTime")),
                Integer.parseInt((String) System.getProperty("maxSize")),
                Integer.parseInt((String) System.getProperty("threadpool.minSize")),
                Integer.parseInt((String) System.getProperty("threadpool.queueSize")));
    }

    @Bean(name="Dispatcher")
    public CreatDispatcher creatDispatcher() {
        Dispatcher dispatcher = new Dispatcher(Integer.parseInt((String) System.getProperty("blockingqueue.size")));
        Thread thread = new Thread(creatDispatcher);
        thread.setName("create.dispatcher");
        thread.start();
        return dispatcher;
    }
    
    @Bean(name="MQLinkDTO")
    public MQLinkDTO creatMQLinkDTO() {
        MQLinkDTO mQLinkDTO = new MQLinkDTO(System.getProperty("topic_Name"),System.getProperty("topic_Key"));
        return mQLinkDTO;
    }
}

Service.java

@Service
 public class Service {
      
      @Autowired
      private Dispatcher dispatcher;
      
      public boolean sendMQ(Object obj){
          return dispatcher.offer(obj);
      }
 }

Controller.java

@RestController
public class Controller {

    @Autowired
    private Service service;
    
    @ApiOperation(value = "获取消息提醒", notes = "获取消息提醒")
    @RequestMapping(value = "/get", method = RequestMethod.POST)
    public Result<NoticeDTO> get(@RequestBody Object obj){
        if(service.sendMQ(obj)){
            return new Result<Object>(obj, ResponseCode.SUCCESS);
        }else{
                    return new Result<Object>(obj, ResponseCode.Fail);
                }    
    }
}

 顺序：JVM首先加载LoadedConfiguration.calss 生成Bean,然后客户端发送请求，经过controller->service->dispatch(调度线程，加入线程池等待就绪)

tip:

1.线程中不支持使用spring的依赖注入

2.在线程中想要获取其他的Bean,需要用到ApplicationContext对象。context.getBean(object.class)

3.做异步处理所以不需要对传入参数加锁，反之需要使用synchronized (obj) {} 对obj加锁,并且使用wait阻塞线程，等待唤醒。

4.使用队列（LinkedBlockingQueue)时，使用offer方法判断是否加入队列，并且在取对象时（take）记得判空。

5.JSON.toJSONString方法可以转Object为Json串,但是会产生转义字符，所以要加入到JSONObject中。