java - Mule ConsumerIterator 中缺少记录

Question

我正在使用 Mule 组件来处理来自 Salesforce 查询的记录。我编写了一个 Mule 组件来使用 ConsumerIterator 以 100 为一组聚合记录。但是，我没有获取所有记录。我包含了一个简单的计数来计算所有记录，并且使用 AtomicInteger 来计算传递给 handOff 方法的总值。我还打印出线程名称，因为我不认为这是线程问题。我正在使用同步流。我正在使用 quartz 组件，但我已将其配置为仅使用 1 个接收器线程。这应该确保我使用相同的线程进行处理。我已经包含了代码。线程名称始终相同。 ConsumerIterator 报告 7348 条记录，“count”变量也报告 7348 条。然而，AtomicInteger 计数器报告 7276 条记录，即少了 72 条记录。请有人看看这个并告诉我原因以及如何解决它？

 public class Aggregator implements Callable {

   private Logger LOG=LoggerFactory.getLogger(Aggregator.class);

    private AtomicInteger at=new AtomicInteger();

  public Aggregator() {
  }

   @Override
   public Object onCall(MuleEventContext context) throws Exception {
    Object obj=context.getMessage().getPayload();
    ConsumerIterator<HashMap<String,Object>> iterator=
               (ConsumerIterator<HashMap<String,Object>>)obj;
    List<HashMap<String,Object>> l=new ArrayList<>();
    LOG.info("Iterator has "+iterator.size()+" elements!");
    int count=0;
    while(iterator.hasNext()){
    count+=1;
        Object payload=iterator.next();
        LOG.info("Processing Next:: => "+payload);
        if(l.size()<100){
            l.add((HashMap<String,Object>)payload);
        }else {

             LOG.info("Just handing off to write :: => Size is "+l.size());
             handoff(l);
             l=new ArrayList<HashMap<String,Object>>();
        }
            if(!iterator.hasNext() && l.size()<=100){
            LOG.info("Now it's less than  100,  and it also has no more 

          elements, so handing off, or they will be lost!! : => 
                             "+l.size()+" 
                                                 elements");
            handoff(l);
            l=new ArrayList<HashMap<String,Object>>();
        }
    }

LOG.info("Done with List !!!!");
    LOG.info("Simple count is "+count);
    LOG.info("Number of elements in counter is "+at.get());
    LOG.info("Thread Name is "+Thread.currentThread());
    return NullPayload.getInstance();

}

private void handoff(List<HashMap<String, Object>> l) {
          at.getAndAdd(l.size());
          LOG.info("handoff:: Thread Name is "+Thread.currentThread());
          LOG.info("Number of elements in List is "+l.size());
          for(HashMap<String, Object> m : l){
             Object obj=m.get("CreatedBy") ;
             if(obj!=null){
                 LOG.info("OBJ====="+obj);
                 LOG.info(obj.getClass().getCanonicalName());
             }
          }

    }

Answer 1

你累了吗CopyOnWriteArrayList(List接口(interface)的线程安全实现可能会有所帮助)尝试更换

List<HashMap<String,Object>> l=new ArrayList<>();

与

List<HashMap<String,Object>> l=new CopyOnWriteArrayList<HashMap<String,Object>>();

和

l=new ArrayList<HashMap<String,Object>>();

与

l=new CopyOnWriteArrayList<HashMap<String,Object>>();