多线程发送微信订阅消息

柔光的暖阳◎ 2023-10-13 20:41 3阅读 0赞

真实代码肯定不能发出来。模拟代码如下：复制/根据业务修改即用

### 1.es.submit 有返回值 ###

List<UserRemindRecord> userList = new ArrayList<>();
        List<UserRemindRecord> list = new CopyOnWriteArrayList<>();
    //原先用的ArrayList 发现获取的数据有问题 并发操作ArrayList不安全啊。！因为你list.add 操作不安全
        List<Long> setList = new CopyOnWriteArrayList<>();
        ResponseResult<WeixinMessageSubscribeSendOneRsp> result = new ResponseResult<>();
    
        @BeforeEach
        public void before() {
            long dataCount = 1000;
            // 创建订单数据。模拟已经插入到数据库的订单
            for (long i = 0; i < dataCount; i++) {
                UserRemindRecord orderVO = new UserRemindRecord();
                orderVO.setId(i + 1);
                orderVO.setUserId(i + 1);
                orderVO.setWeiXinUnionid(String.valueOf(i + 1));
                orderVO.setWxTemplateId((i +1)+ "temId");
                //防止电脑太快，导致都是同一个时间，所以加一个数
                orderVO.setCreateTime(Date.from(LocalDateTime.now().plusSeconds(i).atZone(ZoneId.systemDefault()).toInstant()));
                userList.add(orderVO);
            }
            WeixinMessageSubscribeSendOneRsp oneRsp = new WeixinMessageSubscribeSendOneRsp();
            oneRsp.setErrcode(200);
            result.setData(oneRsp);
            log.info("初始化数据================{}", userList.size());
        }
    
    
        @Test
        public void mid() throws InterruptedException, ExecutionException {
    //        WeixinMessageSubscribeSendOneReq req = new WeixinMessageSubscribeSendOneReq();
            //A.CountDownLatch latch = new CountDownLatch(userList.size());
    //A.这样写是为了下列es.execute(() -> {  无返回值，但是仍想要线程结束，再在主线程中执行代码
            Map<String, TemplateData> m = new HashMap<>();
            long startTime = System.currentTimeMillis();
            log.info("开始时间start={}", startTime);
            // 获取当前系统可用的处理器核心数   
            int i = Runtime.getRuntime().availableProcessors();
            log.info("cpu核心数:{}", i);
    //newFixedThreadpool之前用这个，因阿里禁用容易OOM
            ExecutorService es = new ThreadPoolExecutor(i * 2, 200,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(1024), new ThreadPoolExecutor.AbortPolicy());
    
            List<List<UserRemindRecord>> partList = Lists.partition(userList, 400);
            List<Future<UserRemindRecord>> futures = new ArrayList<>();
    
            for (UserRemindRecord item : userList) {
                Future<UserRemindRecord> fut = es.submit(() -> {
                    WeixinMessageSubscribeSendOneReq req = new WeixinMessageSubscribeSendOneReq();
    
                    req.setTemplate_id(item.getWxTemplateId());
                    req.setPage("pages/index/index");//跳转到首页
                    req.setTouser(item.getWeiXinUnionid());
                    req.setData(m);
                    req.setLang("zh_CN");
                    log.info("微信端推送消息数据==={}", req);
                    // ResponseResult<WeixinMessageSubscribeSendOneRsp> result = feignClient.messageSubscribeSend(req);
                    WeixinMessageSubscribeSendOneRsp oneRsp = result.getData();
                    //log.info("推送订阅消息-微信端返回为=={}", oneRsp);
                      //默认发送成功，记录状态
                        if (oneRsp.getErrcode() != 0) {
                            item.setRemindJson(JSONUtil.toJsonStr(oneRsp));
                            list.add(item);
                            setList.add(item.getUserId());
                        }
    //                   A. latch.countDown();
                    return item;
                });
                futures.add(fut);
            }
    //       A.     latch.await();
           for (Future<UserRemindRecord> future : futures) {
    //此处作用！future.get();类似 A. latch.countDown();  只不过让有返回值的线程执行完毕！！！
    // 否则 下列list.size=0.因为多线程，线程执行顺序无法预测，不等子线程执行，主线程就执行了。
    此时list中都没放进去值啊！
                UserRemindRecord one = future.get();
    
            }
            es.shutdown();
            long endTime = System.currentTimeMillis();
            log.info("结束时间end={}", endTime);
            log.info("消耗时间cast={}", endTime - startTime);
            System.out.println(list.size());
            System.out.println(setList.size());

结果：

1.用es.submit，但是无futures.add(fut);

2.用es.submit,也有futures.add(fut)，但是无future.get(); 可理解没循环进行执行get方法

3.用es.excute,但是无 CountDownLatch latch = new CountDownLatch(userList.size());

都导致list.size() setList.size()不正确！

**原因**：1.多线程环境下并发**访问**和**修改** ArrayList 可能会导致数据不一致或出现异常

2.要等待子线程结束，此时操作的资源副本才能一一归位到自己的集合中\[准确性\]

### 2.es.excute 无返回值 但是实现返回值效果 ###

List<UserRemindRecord> userList = new ArrayList<>();
        List<UserRemindRecord> list = new CopyOnWriteArrayList<>();
        List<Long> setList = new CopyOnWriteArrayList<>();
        ResponseResult<WeixinMessageSubscribeSendOneRsp> result = new ResponseResult<>();
    
        @BeforeEach
        public void before() {
            long dataCount = 1000;
            // 创建订单数据。模拟已经插入到数据库的订单
            for (long i = 0; i < dataCount; i++) {
                UserRemindRecord orderVO = new UserRemindRecord();
                orderVO.setId(i + 1);
                orderVO.setUserId(i + 1);
                orderVO.setWeiXinUnionid(String.valueOf(i + 1));
                orderVO.setWxTemplateId((i + 1) + "temId");
                //防止电脑太快，导致都是同一个时间，所以加一个数
                orderVO.setCreateTime(Date.from(LocalDateTime.now().plusSeconds(i).atZone(ZoneId.systemDefault()).toInstant()));
                userList.add(orderVO);
            }
            WeixinMessageSubscribeSendOneRsp oneRsp = new WeixinMessageSubscribeSendOneRsp();
            oneRsp.setErrcode(200);
            result.setData(oneRsp);
            log.info("初始化数据================{}", userList.size());
        }
    
    
        @Test
        public void mid() throws InterruptedException, ExecutionException {
    //        WeixinMessageSubscribeSendOneReq req = new WeixinMessageSubscribeSendOneReq();
            CountDownLatch latch = new CountDownLatch(userList.size());
            Map<String, TemplateData> m = new HashMap<>();
            long startTime = System.currentTimeMillis();
            log.info("开始时间start={}", startTime);
            int i = Runtime.getRuntime().availableProcessors();
            log.info("cpu核心数:{}", i);
            ExecutorService es = new ThreadPoolExecutor(i * 2, 200,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(1024), new ThreadPoolExecutor.AbortPolicy());
    
            List<List<UserRemindRecord>> partList = Lists.partition(userList, 400);
            List<Future<UserRemindRecord>> futures = new ArrayList<>();
            es.submit(() -> {
                for (UserRemindRecord item : userList) {
    
                    WeixinMessageSubscribeSendOneReq req = new WeixinMessageSubscribeSendOneReq();
    
                    req.setTemplate_id(item.getWxTemplateId());
                    req.setPage("pages/index/index");//跳转到首页
                    req.setTouser(item.getWeiXinUnionid());
                    req.setData(m);
                    req.setLang("zh_CN");
                    log.info("微信端推送消息数据==={}", req);
                    WeixinMessageSubscribeSendOneRsp oneRsp = result.getData();
                    try {
                        Thread.sleep(100);//模拟微信端请求消耗时间
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    //log.info("推送订阅消息-微信端返回为=={}", oneRsp);
                    //默认发送成功，记录状态
                    if (oneRsp.getErrcode() != 0) {
                        item.setRemindJson(JSONUtil.toJsonStr(oneRsp));
                        list.add(item);
                        setList.add(item.getUserId());
                    }
                    latch.countDown();
                }
            });
    
            latch.await();
    
            long endTime = System.currentTimeMillis();
            log.info("结束时间end={}", endTime);
            log.info("消耗时间cast={}", endTime - startTime);
            System.out.println(list.size());//看是否丢失数据
            System.out.println(setList.size());//判断返回的数据是否混乱
        }

此处有个改动。1.线程执行与集合内外互换了下2.加了个Thread.sleep(100)模拟微信端处理请求时间

看结果时间：106960 ~107秒

![6145e225c7fa4ffeab8ed1249ea932ae.png][]

线程池只有一个线程再工作—》进行对集合的操作

改为原来集合外层，es.escute内层时 时间：9198 ~9秒 图中touser也不是顺序执行

![8c7e7d6683d24315a515816339a9d7bd.png][]

我的电脑内核：![808724367d044ac3aea0bcb8412155bd.png][]

9\*6\*2=108 ~107秒 啧啧！反证：一个多线程操作，另一个虽然多线程，但是只是一个线程在工作。没利用多线程优势！

不知道线程池在集合外还是集合里：概念混淆！或者说 明白 线程池里干嘛的 执行资源！！！

在集合内，线程池执行集合中**某个资源**！线程池空闲线程去争取**该资源**

在集合外，线程池执行**集合整体资源！争取该整体==跟同步想想没区别**

### **3.关于线程池的拒绝策略问题** ###

> private ExecutorService es = new ThreadPoolExecutor(i \* 2, 2000, 15, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(1024), new ThreadPoolExecutor. **CallerRunsPolicy**());

**重点想说策略问题：**

若线程池中的核心线程数被用完且[阻塞队列][Link 1]已排满，则此时线程池的资源已耗尽，线程池将没有足够的线程资源执行新的任务。为了保证操作系统的安全，线程池将通过拒绝策略处理新添加的线程任务

实际开发中：大致举例：线程最大树数200，队列1000，但是你要发送的实际人数1万人，此时不同策略就会有不同效果！

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td style="width:188px;"> <p>AbortPolicy</p> </td> 
   <td style="width:262px;">超过了资源如1000，抛异常，报错！</td> 
   <td>抛出 RejectedExecutionException 异常&nbsp; 后续9000不执行</td> 
  </tr> 
  <tr> 
   <td style="width:188px;"> <p>DiscardPolicy</p> </td> 
   <td style="width:262px;">超过了资源如1000，抛多余任务</td> 
   <td>9000不执行，但不报错，会丢失数据</td> 
  </tr> 
  <tr> 
   <td style="width:188px;"> <p>DiscardOldestPolicy</p> </td> 
   <td style="width:262px;">超过了资源如1000，抛弃旧任务</td> 
   <td>队列中前面1000个内时间久的丢弃，会丢失数据</td> 
  </tr> 
  <tr> 
   <td style="width:188px;"> <p>CallerRunsPolicy</p> </td> 
   <td style="width:262px;">超过了资源如1000</td> 
   <td>嘿！主线程还能执行</td> 
  </tr> 
  <tr> 
   <td colspan="3" style="width:188px;"> <p>拒绝新提交任务：</p> <p>1.当我们调用 shutdown 等方法关闭线程池后，即便此时可能线程池内部依然有没执行完的任务正在执行，但是由于线程池已经关闭，此时如果再向线程池内提交任务，就会遭到拒绝。<br> 2.线程池没有能力继续处理新提交的任务，也就是工作已经非常饱和的时候。</p> </td> 
  </tr> 
 </tbody> 
</table>

[线程池的五种拒绝策略\_Fighting0531的博客-CSDN博客][Fighting0531_-CSDN] 参考图例好理解

### 4.多线程中ConcurrentModificationException 等异常原因 ###

参考下列文章讲的确实好啊！

[java.util.ConcurrentModificationException 异常问题详解 - snowater - 博客园 (cnblogs.com)][java.util.ConcurrentModificationException _ - snowater - _ _cnblogs.com]

**多线程学的很惭愧。希望有所帮助**

**参考：**[java多线程：使用newFixedThreadPool方法创建指定线程数量的线程池\_线程池创建固定线程数\_zhaoliubao1的博客-CSDN博客][java_newFixedThreadPool_zhaoliubao1_-CSDN]

[多线程快速处理List集合（结合线程池的使用）\_线程池处理list\_凌晨里的无聊人的博客-CSDN博客][List_list_-CSDN]

[6145e225c7fa4ffeab8ed1249ea932ae.png]: https://img-blog.csdnimg.cn/6145e225c7fa4ffeab8ed1249ea932ae.png
[8c7e7d6683d24315a515816339a9d7bd.png]: https://img-blog.csdnimg.cn/8c7e7d6683d24315a515816339a9d7bd.png
[808724367d044ac3aea0bcb8412155bd.png]: https://img-blog.csdnimg.cn/808724367d044ac3aea0bcb8412155bd.png
[Link 1]: https://so.csdn.net/so/search?q=%E9%98%BB%E5%A1%9E%E9%98%9F%E5%88%97&spm=1001.2101.3001.7020
[Fighting0531_-CSDN]: https://blog.csdn.net/weixin_53946630/article/details/117107417
[java.util.ConcurrentModificationException _ - snowater - _ _cnblogs.com]: https://www.cnblogs.com/snowater/p/8024776.html
[java_newFixedThreadPool_zhaoliubao1_-CSDN]: https://blog.csdn.net/zhaoliubao1/article/details/102779764
[List_list_-CSDN]: https://blog.csdn.net/delete_bug/article/details/125736236