Spring Webflux/Netty/MongoDB로 채팅 서버 구현

• 📌 Goal: Spring Webflux와 MongoDB를 활용해 양방향으로 소통할 수 있는 채팅 서버를 구현한다

1. 환경설정

• MongoDB
  • 외부에서도 DB에 접근할 수 있도록 cloud cluster를 하나 구성하고, Database와 collection을 생성한다. RDBMS에서의 Schema와 Table이라고 생각하면 된다.
  
  
  • MongoDB에서 connection string을 제공하므로 복사해두고, 추후 yml 파일 설정 시에 사용한다.

  mongodb+srv://<username>:<password>@<cluster-address>/<database-name>?retryWrites=true&w=majority
  

  
• Spring Boot
  • Reactive-web과 MongoDB를 사용할 수 있도록 Gradle을 추가해주었다.
    • Reactive web에서는 default 서버로 netty를 사용한다. 따라서 추가적으로 설정할 부분은 없다.

  implementation 'org.springframework.boot:spring-boot-starter-data-mongodb-reactive'
  implementation 'org.springframework.boot:spring-boot-starter-webflux'
  compileOnly 'org.projectlombok:lombok'
  developmentOnly 'org.springframework.boot:spring-boot-devtools'
  annotationProcessor 'org.projectlombok:lombok'
  testImplementation 'org.springframework.boot:spring-boot-starter-test'
  testImplementation 'io.projectreactor:reactor-test'

  • applications.yml 파일에 connection string을 저장해두고 spring boot에서 mongoDB로 바로 접근할 수 있게 한다.

  spring:
  	data:
  	    mongodb:
  	      uri: <connection string>
  

  • 이렇게 구성해둔 설정정보를 통해 추후에 spring boot에서 mongoTemplate를 사용하여 데이터를 CRUD할 수 있게 된다.

 

2. Netty란?

• 우선 Spring 5.X framework에서 사용하는 두가지의 web stack에 대해서 알아보자.
• Spring Framework5는 Servlet Stack, Reactive Stack이라는 두 가지 웹 스택을 제공한다.
  • 하나는 대부분의 Java 엔터프라이즈 애플리케이션이 사용하는 차단 I/O가 있는 고전적인 'Servlet Stack' 이다. Servlet Stack은 Spring MVC 및 Spring Data로 구성되어 있으며, Tomcat, Jetty, Servlet Container에서 동작한다.
  • 다른 하나는 이벤트 루프, 비차단 실행 모델을 기반으로 하여 더 적은 하드웨어 리소스로 높은 동시성을 처리할 수 있는 'Reactive Stack' 으로, Reactive Stack은 Spring WebFlux와 Spring Data 반응형 저장소(Reactive Repositories)를 활용한다. Reactive Stack은 Tomcat, Jetty 외에도 Servlet 3.1+ Container와 Netty, Undertow 같은 논블로킹 서버 위에서 구동된다.

• 기본적으로 Servlet 기반의 스프링은 사용자가 request를 할 때마다 스레드가 만들어진다. 
  • 따라서 여러 사용자가 동시에 요청을 하게 되면 스레드가 계속 생성되면서 개수가 늘어나게 되고, 서버에 과부하가 걸리게 된다.
  • 따라서 Thread 기반 서버들은 서버 하나가 시간을 쪼개서(time slicing) 왔다갔다 하며(context switching) 여러가지 일을 하게 되는데, 이가 자주 발생하면 웹 서버가 느려지는 경향이 있다.
• 하지만 Netty는 비동기 서버이므로, 상당히 빠르다. 
  • 만약 한 사용자가 서버에 요청을 하고 , 이 서버가 DB로 쿼리를 날려 응답이 돌아오기까지 3초정도가 걸린다고 가정했을때, 다른 사용자의 새로운 요청이 들어온다면 A의 응답을 기다리는 동안 B의 요청을 처리해준다. 
  • 이렇게 기다리는 시간동안 할 수 있는 일을 찾아서 비동기적으로 task를 처리하게 된다. 즉 비동기 서버는 싱글 스레드 방식을 사용하는 대신 idle한 시간이 없고, 매우 빨라지게 된다.
  • 하지만 이 서버를 사용하려면 역시 같은 방식으로 동작하는 비동기 DB를 사용해야 한다.

 

3. Tailable - Repository & Model 구현 

@Data
@Document(collection="chat-app")
public class Chat {
    @Id
    private String id;
    private String msg;
    private String sender;
    private String receiver;
    private LocalDateTime createdAt;
}

• Chat model을 구현한다. MongoDB에는 document 형식으로 데이터를 삽입한다.

public interface ChatRepository extends ReactiveMongoRepository {
    @Tailable
    @Query("{sender:?0,receiver:?1}") //해당 쿼리가 동작하게 됨.
    Flux<Chat> mFindBySender(String sender, String receiver);
    //Flux- 데이터의 흐름,끊기지 않고 데이터를 지속적으로 받겠다는 의미
    
    @Tailable
    @Query("{ roomNum: ?0 }") //방 단위로 조회한다
    Flux<Chat> mFindByRoomNum(Integer roomNum);
}

• client가 controller를 통해서 sender가 나인 데이터를 찾는 Tailable Query를 요청했을때, 다른 클라이언트가 sender가 나인 데이터를 DB에 새로 삽입했다고 가정하자.
• 여기에 커서를 닫지 않고 계속 유지하고 최신 DB 내용을 반환하게 되는 @Tailable 어노테이션을 추가하면 Flux를 통해서 데이터가 계속 흘러들어오게 되므로 새로 추가된 데이터까지 반환이 된다.

 

4. SSE Protocol- Controller 구현

• SSE 프로토콜은 Server-Sent Events의 약어로, 1회성으로 데이터를 주고받은 뒤 연결을 종료하는 일반적인 HTTP와는 다르게 최초 연결 이후에 서버의 데이터를 지속적으로, 실시간으로 streaming할 수 있게 된다. (서버가 일방적으로 데이터를 송신하는 단방향 통신) SSE는 기존 HTTP 서버에서 HTTP API 만으로 동작되며, 구현도 Websocket보다 간단하다. 또한 Topic을 정하고 데이터를 구독할 수 있다.
• 주로 Server에서 알람을 푸시하거나, SNS등에서 새로운 피드나 게시물을 받아올 때 사용한다.

+) Websocket vs Server-Side-Event

https://surviveasdev.tistory.com/m/entry/%EC%9B%B9%EC%86%8C%EC%BC%93-%EA%B3%BC-SSEServer-Sent-Event-%EC%B0%A8%EC%9D%B4%EC%A0%90-%EC%95%8C%EC%95%84%EB%B3%B4%EA%B3%A0-%EC%82%AC%EC%9A%A9%ED%95%B4%EB%B3%B4%EA%B8%B0

웹소켓 과 SSE(Server-Sent-Event) 차이점 알아보고 사용해보기

 

• SSE는 기존의 웹 서버 통신 방식을 그대로 사용하기 때문에 구현하기 위해 따로 설정할 것이 없다. 따라서 서버쪽에서는 SSE를 사용하겠다는 헤더만을 설정하면 되며, 이는 MediaType을 text_event_stream_value로 지정해주면 된다.
• Controller는 Restful하게 구현하였다.

@RequiredArgsConstructor
@RestController //데이터 리턴 서버
public class ChatController {
    private final ChatRepository chatRepository;
	
    @CrossOrigin
    @GetMapping(value="/sender/{sender}/receiver/{receiver}",produces= MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<Chat> getMsg(@PathVariable String sender,
                             @PathVariable String receiver){
        return chatRepository.mFindBySender(sender,receiver)
                .subscribeOn(Schedulers.boundedElastic());
    }

    @PostMapping("/chat")
    public Mono<Chat> setMsg(@RequestBody Chat chat){
        chat.setCreatedAt(LocalDateTime.now());
        return chatRepository.save(chat);
    }
    
    @CrossOrigin //단체 채팅방 구현
	@GetMapping(value = "/app/chats/chatrooms/{roomNum}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	public Flux<Chat> findByRoomNum(@PathVariable Integer roomNum) {
		return chatRepository.mFindByRoomNum(roomNum)
				.subscribeOn(Schedulers.boundedElastic());
	}
}

• /chat: 채팅을 전송하면 이를 자동으로 MongoDB에 삽입한다.

• /sender/{name}/receiver/{name}: 해당 url로 접근하면 path variable의 sender/receiver에 따라서 송수신자가 명확히 구분된 메세지를 조회할 수 있고, 또한 서버는 멈추지 않고 계속 새로 삽입되는 데이터를 받아오게 된다.

• /chat/roomNum/{roomNum} : 해당 url로 접근하면 채팅방 전체를 수신자로 하여 채팅이 저장되며, 누가 보냈는지만 정확히 명시하면 된다. 주로 단체 채팅방을 구현할 때 사용된다.
  
  

 

5. 채팅방 뷰 구현 

• Api 구현 만으로는 채팅이 제대로 이루어지고 있는지 확인이 어려웠기 때문에 백엔드 파트에서 채팅방 뷰까지 구현하고, api를 연동하면서 테스트를 진행하였다.
• 채팅방 index, 나의 index, 이름, 상대방 이름 을 url parameter에 담아 request하면 데이터를 읽어오는 api를 적용하여 이전 채팅 내역을 조회하고, 나와 상대방을 분별하며, 누가 어떤 채팅방에 채팅을 보냈는지 명시하여 DB에 채팅을 저장한다.
• 또한 아래 전송창에서 채팅을 보내면 DB에 채팅이 저장되는데, 이때 채팅방을 조회하는 api(Server)에 새롭게 데이터가 흘러들어와 event가 발생하면 채팅창을 초기화하여 새롭게 렌더링해 새로운 채팅까지 뷰에 추가하여 실시간으로 채팅을 확인할 수 있도록 하였다.

const searchParams=new URLSearchParams(location.search);


let userIdx=searchParams.get('user-idx') //유저 인덱스 
let userName=searchParams.get('user-name'); //현재 유저 이름
let roomNum=searchParams.get('room-num') // 채팅방 번호 
let otherName=searchParams.get('other-name')// 상대방 이름
console.log(userIdx)

document.querySelector("#username").innerHTML = otherName;
//SSE 연결 - DB에 새로운 채팅 내역이 들어올때마다 event 발생 (초기화)
let eventSite= "http://{aws ip 주소}" //spring application이 돌아가는 주소 ex) "http://localhost:9001"
const eventSource= new EventSource(`${eventSite}/chatrooms/${roomNum}`); 

 // 계속 데이터가 흘러들어오므로 postman에서 테스트 X, 그냥 url 그대로 접속



eventSource.onmessage=(event)=>{
    const data=JSON.parse(event.data);
    //상대방으로부터 메세지가 전송되는 이벤트가 발생하면 DB에 저장된 데이터를 불러온다.
    if(data.sender_idx==userIdx){ //전송자가 내가 아니면 다른사람이므로 반대쪽에 렌더링하면 됨
        //파란박스 (내가 보낸 메세지)
        initMyMessage(data);
    }
    else{
        //회색박스 (상대방이 보낸 메세지)
        initYourMessage(data);
    }
    

}

function getSendMsgBox(data) {

    let md = data.createdAt.substring(5, 10)
	let tm = data.createdAt.substring(11, 16)
	convertTime = tm + " | " + md
    //내가 입력한 채팅 박스 생성하기
    return `
    <div class="sent_msg">
    <p>${data.msg}</p>
    <span class="time_date"> ${convertTime} / <b>${data.sender_name}</b> </span>
    </div>
    `;
}

function getReceivedMsgBox(data) {

    let md = data.createdAt.substring(5, 10)
	let tm = data.createdAt.substring(11, 16)
	convertTime = tm + " | " + md
    //상대편에서 보낸 채팅 박스 생성하기
    return `
    <div class="received_withd_msg">
    <p>${data.msg}</p>
    <span class="time_date"> ${convertTime} / <b>${data.sender_name}</b></span>
    </div>
    `;
}

//최초 채팅방 로드시 이전 DB 저장 내역 불러오며 초기화

function initMyMessage(data){
    //내가 보낸 메세지 초기화
    let chatBox= document.querySelector("#chat-box");
   
   let chatOutGoingBox =document.createElement("div");
   chatOutGoingBox.className="outgoing_msg";
   
   chatOutGoingBox.innerHTML=getSendMsgBox(data);
    chatBox.append(chatOutGoingBox);
    document.documentElement.scrollTop = document.body.scrollHeight;
  
}

function initYourMessage(data){
    //상대 쪽으로부터 메세지가 전송되면 DB로부터 불러옴
    let chatBox= document.querySelector("#chat-box");
   
   let chatReceivedBox =document.createElement("div");
   chatReceivedBox.className="received_msg";
  
   chatReceivedBox.innerHTML=getReceivedMsgBox(data);
    chatBox.append(chatReceivedBox);
    document.documentElement.scrollTop = document.body.scrollHeight;
}


//AJAX로 채팅 메세지 전송

async function newChat(){
    //DB에 insert하면 자동으로 event가 발생하면서 chatroom의 다른 상대방에게 보내짐 (다중 채팅도 가능)
  
    let msgInput=document.querySelector("#chat-outgoing-msg");

   
   let date=new Date();
   let now= date.getHours()+":"+date.getMinutes()+" | "+date.getMonth()+"- "+date.getDate();

   let chat={
       sender_idx:userIdx,
       sender_name:userName,
       room_num:roomNum,
       msg: msgInput.value
   };

    fetch(`${eventSite}`,{
       method:"post",//http post 메소드 (새로운 데이터를 write할때 사용)
       body:JSON.stringify(chat),
       headers:{
           "Content-Type":"application/json; charset=utf-8"
       }

   });

    msgInput.value="";
}

//채팅 전송 버튼 누를 시
document.querySelector("#chat-send").addEventListener("click",()=>{
   newChat();
})


//enter 누를 시
document.querySelector("#chat-outgoing-msg").addEventListener("keydown",(e)=>{
   if(e.keyCode==13){
       newChat();
   }
})

 

- 전체 코드는 하단 깃허브 주소 참고

https://github.com/ashlovesliitea/silverlining-chatapp

GitHub - ashlovesliitea/silverlining-chatapp

 

 

🙏 참고 자료

• https://github.com/codingspecialist/springboot-webflux-mongo-chatapp

GitHub - codingspecialist/springboot-webflux-mongo-chatapp

• https://sjh836.tistory.com/181

Spring Event와 SSE 로 리액티브하게 접근하기 (EventListener, Server-Sent Events, 비동기 컨트롤러, RxJava로