解决方案是使用计时器和timertask（作为内部类）。如果接收到所有相关事件，则构造函数将创建计时器并按所需的秒数调度timertask，然后调用回调并取消计时器。在少有的情况下，如果事件丢失，timertask会启动并向回调发送一组不完整的事件并记录错误。
仅供参考。
公共类事件集{

private MonitorEventsConsumer eventsConsumer;
// For timeout handling
private Timer timer;
private int waitTime; // seconds to wait 
protected boolean isComplete;
private String localTxnId;
String eventSourceAddress;
private boolean txnEndReceived;
private int maxCounter = 0;
private List<String> monEvents = new ArrayList<>();
public EventSet(String localTxnId) {
    this(localTxnId, 20); // Default the waiting time for all events to be received to 20 secs
}
public EventSet(String localTxnId, int waitTime) {
    this.localTxnId = localTxnId;
    timer = new Timer();
    this.waitTime = waitTime;
    timer.schedule(new TimeoutTask(), waitTime*1000);
}
public boolean isNew() {
    return monEvents.isEmpty();
}
public void addEvent(String eventAsXML) {
    monEvents.add(eventAsXML);
    String counter = StringUtils.substringBetween(eventAsXML, "wmb:counter=\"", "\"/>");
    eventSourceAddress = StringUtils.substringBetween(eventAsXML, " wmb:eventSourceAddress=\"", "\">");
    if (eventSourceAddress.endsWith("transaction.End")) {
        txnEndReceived = true;
    }
    int iCounter = Integer.parseInt(counter);
    if (iCounter > maxCounter) {
        maxCounter = iCounter;
    }
    if (txnEndReceived && maxCounter == monEvents.size()) {
        // all events should have been received unless maxCounter = 1
        // Call back to the onMessageHandler to process this event set.
        eventsConsumer.sendEventsToNjams(localTxnId, this);
        isComplete = true;
        timer.cancel();
    }
}
public List<String> getMonEvents() {
    return monEvents;
}
public void clear() {
    monEvents.clear();
    localTxnId = null;
    isComplete = false;
    maxCounter = 0;
}
public void registerHandler(final MonitorEventsConsumer eventsConsumer) {
    if (this.eventsConsumer == null) {
        this.eventsConsumer = eventsConsumer;
    }
}
class TimeoutTask extends TimerTask {
    public void run() {            
        eventsConsumer.sendEventsToNjams(localTxnId, EventSet.this);
        isComplete = true;
        System.err.println(String.format("Message events for local txn id '%s' were incomplete after waiting for %d seconds. they were sent to nJAMS in this state", localTxnId, waitTime));
        timer.cancel(); //Terminate the timer thread
    }
}

}

我不知道你的库是否支持这种方法，但是一个普通的java尝试会像这样：

public class MessageProcessor extends Thread {
  private static final long PROCESSOR_TIMEOUT   = 3600 * 1000 /* 1 hour? */;
  private static final long ONE_MESSAGE_TIMEOUT = 1000 /* 1 sec? */;
  private long              m_startTime         = 0;
  public void startProcessing() {
    m_startTime = System.currentTimeMillis();
    start();
  }
  @Override
  public void run() {
    while(true) {
      // ---------------------------------
      // check abort condition
      // ---------------------------------
      if(System.currentTimeMillis() - m_startTime > PROCESSOR_TIMEOUT) {
        break;
      }
      // ---------------------------------
      // handle 1 message (with timeout)
      // ---------------------------------
      try {
        // pseudo code
        collectOneMessage(ONE_MESSAGE_TIMEOUT);
      }
      catch(InterruptedException e) {
        // ignoring timeout exception
      }
    }
    // ---------------------------------
    // finish up
    // ---------------------------------
    // ...
  }
}