ddakker님의 블로그

서버 모니터링 시스템 구축 중 Tomcat Comet으로 했던걸 WebSocket로 변경 하였다.

단순함!

WebSocket Servlet 에서는 Client Open/Close 만 관리하는 public static Set 객체에 Client 정보를 관리하고, Queue Receive 에서 해당 Set 객체에 접근 하여 각각의 Client 에 데이터를 Push 한다.

Thread 간 통신에 public static 변수를 사용하는게 좋은건지 모르겠으나.. Thread 관련 지식이 부족한 관계로 현재는 우선 패스한다.

import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


@ServerEndpoint(value = "/ws/was")
public class WasWebSocket {

	private static Logger log = LoggerFactory.getLogger(WasWebSocket.class);
	

    public static Set connections = new CopyOnWriteArraySet<>();

    private Session session;
    
    public WasWebSocket() {
        log.info("----------- thread start");
	}

    public Session getSession() {
    	return this.session;
    }

    @OnOpen
    public void onOpen(Session session) {
    	log.info("onOpen this: {}, this.sessson: {}", this.getClass(), session);
        this.session = session;
        synchronized (connections){
        	connections.add(this);
        }
    }


    @OnClose
    public void onClose() {
    	log.info("onClose");
    	synchronized (connections){
    		connections.remove(this);
    	}
    }


    @OnMessage
    public void onMessage(String message) {
    	log.info("@OnMessage");
    	log.info("websocket message: " + message);
    }



    @OnError
    public void onError(Throwable t) throws Throwable {
        log.error("e: ", t.toString());
        synchronized (connections){
    		connections.remove(this);
    	}
    }
}

function wasStatusListen() {
	var oSocket = new WebSocket(WS_URL + "/ws/was");
	 
    oSocket.onmessage = function (e) { 
        var json = eval("(" + e.data + ")");
		if (json.result && json.result == "0000") {
			var wasNode = json.data.name;
			var heapUsedPercent = json.data.heapUsedPercent;
			..
			..
			처리
			
		}
    };
 
    oSocket.onopen = function (e) {
        console.log("open");
    };
 
    oSocket.onclose = function (e) {
        console.log("close");
        wasStatusListen();
    };
    
    window.unload = function() {
		if (oSocket.readyState != 3)
			oSocket.disconnect();
	}
}
wasStatusListen();

서버 모니터링 시스템을 만들고 있다.

구조는 서버의 MBean 및 Background 테스팅 도구가 실행한 정보를 각각의 N대의 서버에서 Message Qeue에 전달 하고, 모니터링 WAS에서 Queue 를 Receive 하여 Tomcat Long Pooling 을 활용하여 N대의 Client Browser 에 전송 해 준다.

public static List connections = new ArrayList<>(); 부분이 접속된 Client Browser 정보이므로, Queue Receive Event 시점에 connections Client 갯수만큼 PrintWriter 활용 해서 write 하고, connections 의 CometEvent 객체를 Close 하고, connections에서 Remove() 시킨다.

접속된 Client 정보를 public static 변수로 다른 Thread 에서 제어 하는게 잘 하는건지는 모르겠지만. 잘 된다...

사실 처음에 Tomcat Comet 방식을 이용했지만 지금은 WebSocket 방식으로 동일 하게 활용중이다.

다음 글은 동일한 상황에서 Comet 를 Websocket로 변환.. 거의 동일함..

import java.io.IOException;
import java.io.PrintWriter;
import java.util.ArrayList;
import java.util.List;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.catalina.comet.CometEvent;
import org.apache.catalina.comet.CometProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


@WebServlet("/comet/testCase")
public class TestCaseServlet extends HttpServlet implements CometProcessor {
	private static Logger log = LoggerFactory.getLogger(TestCaseServlet.class);
	
	public static List connections = new ArrayList<>();

	public void init() throws ServletException {
        log.info("----------- thread start");
    }

    public void destroy() {
    	log.info("---------- thread stop");
    	
    	synchronized (connections) {
			connections.clear();
		}
    }

	@Override
	public void event(CometEvent event) throws IOException, ServletException {
		log.info("testCase event: " + event.getEventType());
		HttpServletRequest request = event.getHttpServletRequest();
		HttpServletResponse response = event.getHttpServletResponse();
		response.setCharacterEncoding("UTF-8");
		event.setTimeout(1000*60*60);
		
		if (event.getEventType() == CometEvent.EventType.BEGIN) {
            log.info("Begin for request: {}, response: {}, session: {}", request, response, request.getSession(true).getId());
            synchronized(connections) {
            	connections.add(event);
            }
        } else if (event.getEventType() == CometEvent.EventType.ERROR) {
            log.info("Error for session: {}", request.getSession(true).getId());
            synchronized(connections) {
            	connections.remove(event);
            }
            PrintWriter writer = response.getWriter();
            writer.println("{result: '9999', msg: 'timeout'}");
            event.close();
        } else if (event.getEventType() == CometEvent.EventType.END) {
            log.info("End for session: " + request.getSession(true).getId());
            synchronized(connections) {
            	connections.remove(event);
            }
            PrintWriter writer = response.getWriter();
            writer.println("{result: '0000', msg: 'end'}");
            event.close();
        } else if (event.getEventType() == CometEvent.EventType.READ) {
        	 log.info("read");
            while (true) {
            	;
            }
        }
	}
}