微信路由消息机制--------线程池使用

ゞ浴缸里的玫瑰 2022-02-22 00:37 408阅读 0赞

**线程池接口： **

//接口
    public interface Executor {
    
        void execute(Runnable command);
    }
    ----------------------------------
    //使用
    ExecutorService executorService = Executors.newFixedThreadPool(10);  
      
    executorService.execute(new Runnable() {  
        public void run() {  
            System.out.println("Asynchronous task");  
        }  
    });  
    //关闭  
    executorService.shutdown();

**任务的委托（Task Delegation）**

下方展示了一个线程的把任务委托异步执行的ExecutorService的示意图。

![165710_YOvh_136226.png][]

壹旦线程把任务委托给 ExecutorService，该线程就会继续执行与运行任务无关的其它任务。

这里有几种不同的方式让你将任务委托给壹個 ExecutorService：

execute(Runnable)  这种方式没有办法获取执行 Runnable 之后的结果
    submit(Runnable)   同样接收壹個 Runnable 的实现作为参数，但是会返回壹個 Future 对象。可以用于判断 Runnable 是否结束执行。
    submit(Callable)  
    invokeAny(...)  
    invokeAll(...)

**在微信后台中使用：**

**   1.创建线程池**

this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE);

2.接收线程结果

final List<Future<?>> futures = new ArrayList<>();
        for (final WxMpMessageRouterRule rule : matchRules) {
          // 返回最后一个非异步的rule的执行结果
          if (rule.isAsync()) {
            futures.add(
              this.executorService.submit(new Runnable() {
                @Override
                public void run() {
                  rule.service(wxMessage, context, mpService, WxMpMessageRouter.this.sessionManager, WxMpMessageRouter.this.exceptionHandler);
                }
              })
            );
          } else {
           //返回同步最后一个，每次res赋值，保留最后一个，所有通用写最后
            res = rule.service(wxMessage, context, mpService, this.sessionManager, this.exceptionHandler);
            // 在同步操作结束，session访问结束
            this.log.debug("End session access: async=false, sessionId={}", wxMessage.getFromUser());
            sessionEndAccess(wxMessage);
          }

**Future模式：**我有一个任务，提交给了Future，Future替我完成这个任务。期间我自己可以去做任何想做的事情。一段时间之后，我就便可以从Future那儿取出结果。  
**Future接口提供方法来检测任务是否被执行完，等待任务执行完获得结果，也可以设置任务执行的超时时间。**

**V get () :**等待任务执行结束，然后获得V类型的结果。

if (futures.size() > 0) {
          this.executorService.submit(new Runnable() {
            @Override
            public void run() {
              for (Future<?> future : futures) {
                try {
                  future.get();
                  WxMpMessageRouter.this.log.debug("End session access: async=true, sessionId={}", wxMessage.getFromUser());
                  // 异步操作结束，session访问结束
                  sessionEndAccess(wxMessage);
                } catch (InterruptedException e) {
                  WxMpMessageRouter.this.log.error("Error happened when wait task finish", e);
                  Thread.currentThread().interrupt();
                } catch (ExecutionException e) {
                  WxMpMessageRouter.this.log.error("Error happened when wait task finish", e);
                }
              }
            }
          });
        }

[165710_YOvh_136226.png]: /images/20220222/5e171426717a4a90b57bec4852d79cbc.png