OpenfirePresenceSpooler

 

ㅁ openfire가 direct presence 라서 모델을 조금 변경함.

 

private AtomicInteger scheduledExecutorServiceCount = new AtomicInteger(); /** * Initializes the spool manager. */ @PostConstruct public void init() throws Exception { logger.info(getClass().getName() + "start OpenfirePresenceSpooler initialize."); queue = queueFactory.getQueue(PresenceQueueFactory.SPOOL); if (logger.isInfoEnabled()) { StringBuffer infoBuffer = new StringBuffer(64).append(getClass().getName()).append(" uses ").append(numThreads).append(" Thread(s)"); logger.info(infoBuffer.toString()); } active.set(true); workerService = JMXEnabledThreadPoolExecutor.newFixedThreadPool("com.xxxxx.openfire:type=component,component=pushcontainer,name=presencespooler,sub-type=threadpool", "spooler", numThreads); dequeueService = JMXEnabledThreadPoolExecutor.newFixedThreadPool("com.xxxxx.openfire:type=component,component=pushcontainer,name=presencespooler,sub-type=threadpool", "dequeuer", numDequeueThreads); ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(() -> { try { logger.info(Thread.currentThread().getId() + "]------------------start ScheduledExecutorService----------------------"); logger.info(Thread.currentThread().getId() + "]-- execute cnt : " +scheduledExecutorServiceCount.incrementAndGet()); logger.info(Thread.currentThread().getId() + "]-- packetBuffer.size :" + packetBuffer.size()); final List<Presence> temp; synchronized (packetBuffer) { temp = new ArrayList<Presence>(packetBuffer); packetBuffer = new ArrayList<>(); } presenceProcessor.service(temp); logger.info(Thread.currentThread().getId() + "]-- execute cnt : " +scheduledExecutorServiceCount.incrementAndGet() + "------------------end ScheduledExecutorService----------------------"); } catch (Exception e) { } }, 12, 3, TimeUnit.SECONDS); for (int i = 0; i < numDequeueThreads; i++) { Thread reader = new Thread(this, "Dequeue Thread #" + i); dequeueService.execute(reader); }