應用與紀事隊列存儲器不斷生長

我已經實現其接收網絡消息的簡單的Spring啓動應用程序，使用排隊成SingleChronicleQueue appender.writeText（STR），使用tailer.readText用於消息另一個線程輪詢（）。經過一些處理後，處理後的消息被放置在另一個SingleChronicleQueue中發送出去。我在應用程序中有三個隊列。應用與紀事隊列存儲器不斷生長

應用旋轉每天晚上文件和第一奇怪的是，文件大小（每個Q）是相同的（不同的每一個Q）。最大的cq4文件大約是每天220MB。

，我所面臨的問題是，在從開頭有三個天，直至現在內存從480MB增長到1.6GB，它只是不合理的。

我有我缺少一些組態一個概念，還是我的一個天真/壞實施。（我應該在每次使用後都不關閉appender和tailer）。

這裏是一個剝離下來的例子，也許有人可以提供一些線索。

@Service 
public class QueuesService { 
    private static Logger LOG = LoggerFactory.getLogger(QueuesService.class); 

    @Autowired 
    AppConfiguration conf; 

    private SingleChronicleQueue Q = null; 
    private ExcerptAppender QAppender = null; 
    private ExcerptTailer QTailer = null; 

    public QueuesService() { 
    } 

    @PostConstruct 
    private void init() { 

     Q = SingleChronicleQueueBuilder.binary(conf.getQueuePath()).indexSpacing(1).build(); 
     QAppender = Q.acquireAppender(); 
     QTailer = Q.createTailer(); 
    } 

    public ExcerptAppender getQAppender() { 
     return QAppender; 
    } 

    public ExcerptTailer getQTailer() { 
     return QTailer; 
    } 
} 


@Service 
public class ProcessingService { 
    private static Logger LOG = LoggerFactory.getLogger(ProcessingService.class); 

    @Autowired 
    AppConfiguration conf; 

    @Autowired 
    private TaskExecutor taskExecutor; 

    @Autowired 
    private QueuesService queueService; 

    private QueueProcessor processor = null; 

    public ProcessingService() { 
    } 

    @PostConstruct 
    private void init() { 
     processor = new QueueProcessor(); 
     processor.start(); 
    } 

    @Override 
    public Message processMessage(Message msg, Map<String, Object> metadata) throws SomeException { 

     String strMsg = msg.getMessage().toString(); 

     if (LOG.isInfoEnabled()) { 
      LOG.info("\n" + strMsg); 
     } 

     try { 
      queueService.getQAppender().writeText(strMsg); 

      if (LOG.isInfoEnabled()) { 
       LOG.info("Added new message to queue. index: " + queueService.getQAppender().lastIndexAppended()); 
      } 
     } 
     catch(Exception e) { 
      LOG.error("Unkbown error. reason: " + e.getMessage(), e); 
     } 
    } 

    class QueueProcessor extends Thread { 

     public void run() { 
      while (!interrupted()) { 
       try { 
        String msg = queueService.getEpicQTailer().readText(); 

        if (msg != null) { 
         long index = queueService.getEpicQTailer().index(); 
         // process 
        } 
        else { 
         Thread.sleep(10); 
        } 
       } 
       catch (InterruptedException e) { 
        LOG.warn(e); 
        this.interrupt(); 
        break; 
       } 
      } 

      ThreadPoolTaskExecutor tp = (ThreadPoolTaskExecutor) taskExecutor; 
      tp.shutdown(); 
     } 
    } 
}

來源

2017-02-03 Gal Nitzan