채팅서버 JobQueue 방식으로 부하 줄이기(command 패턴)

 

command 패턴으로 서버에서 처리할 명령을 바로 처리하지 않고 우선 처리할 명령을 Queue 에 먾어 넣어 모아두었다가

이 방식은 서버가 처리 할수 있을때 처리 하는 방식으로 직접적으로 명령을 다수의 스레드가 한번에 처리 할때 lock 을 걸어야 하는 상황에서의 부하를 줄일 수 있다

---

 

설명

이전까지는 클라이언트 세션을 연결되자마자 바로 GameRoom의 List에 넣어주고 있었지만, 이번에는 JobQueue 개념을 이용해 Queue에 일단 GameRoom의 List에 넣어주는 작업을 '예약'해주고 있다.

---

코드

ServerCore

JobQueue.cs

public interface IJobQueue
    {
        void Push(Action job);
    }

    public class JobQueue : IJobQueue
    {
        Queue<Action> _jobQueue = new Queue<Action>();  
        object _lock = new object();
        bool _flush = false; // 큐에 쌓인걸 '자신이' 실행할건지. 누군가 하고 있으면 자신은 하지 않는다.

        public void Push(Action job)
        {
            bool flush = false;

            lock(_lock)
            {
                _jobQueue.Enqueue(job);
                if (_flush == false)
                    flush = _flush = true;

            }

            if (flush)
                Flush();
        }

        void Flush()
        {
            while(true)
            {
                // 하나씩 꺼내는 와중에도 다른 애가 Push해서 JobQueue에다 넣을 수 있기 때문에 Pop을 할 때 lock을 잡아줘야 한다.
                Action action = Pop();
                if (action == null)
                    return;

                action.Invoke();
            }
        }

        Action Pop()
        {
            lock(_lock)
            {
                if(_jobQueue.Count == 0)
                {
                    _flush = false;
                    return null;
                }
                return _jobQueue.Dequeue(); 
            }
        }
    }

public class Listener
	{
		Socket _listenSocket;
		Func<Session> _sessionFactory;

		public void Init(IPEndPoint endPoint, Func<Session> sessionFactory, int register = 10, int backlog = 10)
		{
			_listenSocket = new Socket(endPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
			_sessionFactory += sessionFactory;

			// 문지기 교육
			_listenSocket.Bind(endPoint);

			// 영업 시작
			// backlog : 최대 대기수
			_listenSocket.Listen(backlog);

			for (int i = 0; i < register; i++)
			{
				SocketAsyncEventArgs args = new SocketAsyncEventArgs();
				args.Completed += new EventHandler<SocketAsyncEventArgs>(OnAcceptCompleted);
				RegisterAccept(args);
			}
		}
        
        ...

Server

GameRoom.cs

 class GameRoom : IJobQueue
    {
        List<ClientSession> _sessions = new List<ClientSession>();
        JobQueue _jobQueue = new JobQueue();

        public void Push(Action job)
        {
            _jobQueue.Push(job);
        }

        public void Broadcast(ClientSession session, string chat)
        {
            // 이 부분은 다른 쓰레드와 공유하고 있지 않음
            S_Chat packet = new S_Chat();
            packet.playerId = session.SessionId;
            packet.chat = $"{chat} 나는 {packet.playerId}";
            ArraySegment<byte> segment = packet.Write();

            foreach (ClientSession s in _sessions)
                s.Send(segment);
            
        }

        public void Enter(ClientSession session)
        {
            _sessions.Add(session);
            session.Room = this;
        }

        public void Leave(ClientSession session)
        {
            _sessions.Remove(session);
        }
    }

ClientSession.cs

class ClientSession : PacketSession
	{
		public int SessionId { get; set; }	
		public GameRoom Room { get; set; }

		public override void OnConnected(EndPoint endPoint)
		{
			Console.WriteLine($"OnConnected : {endPoint}");

			Program.Room.Push(() => Program.Room.Enter(this));
		}

		public override void OnRecvPacket(ArraySegment<byte> buffer)
		{
			PacketManager.Instance.OnRecvPacket(this, buffer);
		}

		public override void OnDisconnected(EndPoint endPoint)
		{
			SessionManager.Inst.Remove(this);
			if(Room != null)
            {
				Room.Push(() => Room.Leave(this));
				Room = null;
            }
			Console.WriteLine($"OnDisconnected : {endPoint}");
		}

		public override void OnSend(int numOfBytes)
		{
			Console.WriteLine($"Transferred bytes: {numOfBytes}");
		}
	}

PacketHandler.cs

class PacketHandler
{
	public static void C_ChatHandler(PacketSession session, IPacket packet)
	{
		C_Chat chatPacket = packet as C_Chat;
		ClientSession clientSession = session as ClientSession;

		if (clientSession.Room == null)
			return;

		GameRoom room = clientSession.Room;

		// 할 일을 바로 해주는게 아니라 Push
		room.Push(() => room.Broadcast(clientSession, chatPacket.chat));
	}
}