flume拦截器开发中CountDownLatch的使用以及线程安全的ArrayList使用

迈不过友情╰ 2021-11-11 22:16 196阅读 0赞

public List<Event> intercept(List<Event> eventList) {
    
            final CountDownLatch latch = new CountDownLatch(eventList.size());
    //        final List<Event> resultList = new ArrayList<Event>(eventList.size());
            // 1
            final List<Event> resultList = Collections.synchronizedList(new ArrayList<Event>(eventList.size()));
    
            // todo 先给list全赋值为null，站位，然后更新给对应下标赋值
            for (int j = 0; j < eventList.size(); j++) {
                resultList.add(null);
            }
    
            for (int j = 0; j < eventList.size(); j++) {
    
                final int index = j;
    
                threadPool.execute(new Runnable() {
    
                    @Override
                    public void run() {
                        try {
                            Event event = intercept(eventList.get(index));
    
                            if (event != null) {
                                resultList.set(index, event);
                            }
    
                        } catch (Exception ex) {
                            logger.error("[{}]线程内异常：{}", Thread.currentThread().getName(), ex);
    
                        } finally {
                            latch.countDown();
    
                        }
                    }
                });
    
            }
            try {
                // 阻塞当前线程直到latch中数值为零才执行
                latch.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
    
            // 剔除List中所有的空元素（null），目的是为了保证数据有序
            resultList.removeAll(Collections.singleton(null));
    
            return resultList;
        }