[식구하자_MSA] 🚀 채팅&알림 마이크로 서비스: 알림 기능 구현(SseEmitter+Redis) - 3편

지난 포스팅에 이어 채팅 마이크로 서비스 알림 기능 과정을 포스팅 하도록 하겠습니다. 주요 코드만 다룰 예정이라 다루지 않는 코드들은 아래 github를 통해 참고하시길 바랍니다🙏🙏(msa-master 브랜치)

👉 참고 코드 : https://github.com/LminWoo99/PlantBackend/tree/msa-master

🤔 알림 기능 구현에 대한 고민

요구사항 분석

식구하자 프로젝트의 알림 기능은 1:1 채팅에서 읽지 않은 메시지에 대해 알림을 제공하는 것이 핵심입니다.

채팅 프로세스 설계

1. 클라이언트: A의 메시지를 서버로 전송
2. 서버: B의 접속 상태 확인
3. 서버: 접속 상태에 따른 readcount 설정 (접속 중: 0, 비접속: 1)
4. 서버: 메시지 정보 추가 후 클라이언트에 전달
5. 클라이언트: 받은 메시지를 서버로 재전송
6. 서버: 메시지 DB 저장 및 필요 시 알림 발송 (readcount가 1일 경우)

핵심 요구사항

1. 실시간 채팅방 접속자 수 파악
2. 사용자의 채팅방 입장 시 접속 정보 관리

🤷‍♂️ 해결 방안 모색

초기 접근: Polling 방식

• 개념: 클라이언트가 주기적으로 서버에 데이터 요청
• 장점: 구현이 간단함
• 단점:
  • 서버 부하 증가 가능성
  • 실시간성과 효율성 사이의 트레이드오프
  • 불필요한 요청으로 인한 리소스 낭비

최종 해결책: Stomp, Redis, SSE 조합

1. Stomp의 ChannelInterceptor 활용

• Stomp 프로토콜의 주요 이벤트(connect, send, disconnect, subscribe) 감지
• ChannelInterceptor 인터페이스 구현을 통한 사용자 정의 로직 추가
  • preSend: 메시지 전송 전 처리 (예: 메시지 유효성 검사)
  • postSend: 메시지 전송 후 처리 (예: 로깅, 통계 업데이트)
  • preReceive: 메시지 수신 전 처리
  • postReceive: 메시지 수신 후 처리
• 채팅방 연결 시 사용자 정보를 Redis에 실시간으로 저장
  • 키: 채팅방 ID, 값: 접속 중인 사용자 목록
  • 사용자 접속/퇴장 시 Redis 데이터 즉시 업데이트

2. Redis 사용 결정 이유

• 채팅 관련 데이터의 특성에 최적화
  • 빈번한 읽기/쓰기 작업 처리에 탁월한 성능
  • 실시간 데이터 처리에 적합한 인메모리 데이터 구조
• 기존 관계형 DB와의 부하 분산
  • 채팅 세션 정보와 같은 휘발성 데이터 관리에 적합
  • 주 데이터베이스의 부하 감소로 전체 시스템 성능 향상
• 다양한 데이터 구조 지원
  • Strings, Lists, Sets, Sorted Sets, Hashes 등 다양한 데이터 타입 제공
  • 채팅방 목록, 사용자 세션 등 다양한 정보를 효율적으로 저장 및 관리

3. SSE(Server-Sent Events) 도입

• 서버에서 클라이언트로의 단방향 실시간 데이터 전송 구현
  • WebSocket과 달리 서버에서 클라이언트로의 단방향 통신만 지원
  • HTTP 프로토콜 사용으로 추가 설정 없이 방화벽 통과 가능
• 알림 기능 구현에 최적화
  • 새 메시지 도착, 읽지 않은 메시지 수 업데이트 등 실시간 알림 제공
  • 클라이언트의 EventSource 객체를 통한 간편한 구현
• 효율적인 리소스 사용
  • 연결 유지에 필요한 리소스가 WebSocket에 비해 적음
  • 자동 재연결 기능 내장으로 안정성 향상
• 브라우저 호환성
  • 대부분의 현대 브라우저에서 지원되어 폭넓은 사용자 환경 커버 가능

1️⃣ SSE를 활용한 알림 기능

컨트롤러 구현

@RestController
@RequestMapping("/notification")
@Slf4j
@RequiredArgsConstructor
public class NotificationController {

    private final NotificationService notificationService;

    /**
     * @title 로그인 한 유저 SSE 연결
     * @param : String lastEventId,String jwtToken
     **/
    @GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseEntity<SseEmitter> connect(
            @RequestHeader(value = "Last-Event-ID", required = false, defaultValue = "") String lastEventId,
            @RequestHeader("Authorization") String jwtToken) {
        SseEmitter emitter = notificationService.subscribe(lastEventId, jwtToken);

        return ResponseEntity.ok(emitter);
    }

    /**
     * @title 로그인 한 유저의 모든 알림 조회
     * @param : Long memberNo(현재 접속한 멤버)
     **/
    @GetMapping("/all")
    public ResponseEntity<List<NotificationResponse>> notifications(Long memberNo) {
        return ResponseEntity.ok(notificationService.findAllById(memberNo));
    }

    /**
     * @title 알림 읽음 상태만 변경
     * @param : Long id(알림 id)
     **/
    @PatchMapping("/checked/{id}")
    public ResponseEntity<StatusResponseDto> readNotification(@PathVariable("id") Long id) {
        notificationService.readNotification(id);
        return ResponseEntity.ok(StatusResponseDto.success());
    }

    /**
     * @title 알림 삭제
     * @param : NotificationDeleteRequest request
     **/
    @DeleteMapping
    public ResponseEntity<StatusResponseDto> deleteNotification(@RequestBody NotificationDeleteRequest request) {
        notificationService.deleteNotification(request.getIdList());
        return ResponseEntity.ok(StatusResponseDto.success());
    }


}

서버에서는 EventSource를 통해 날아오는 요청을 처리할 컨트롤러가 필요하다. sse 통신을 하기 위해서는 MIME 타입을 text/event-stream로 해줘야한다.

추가적으로 Last-Event-ID라는 헤더를 받고 있는 것을 볼 수 있다. 이 헤더는 항상 담겨있는 것은 아니다. 만약 sse 연결이 시간 만료 등의 이유로 끊어졌을 경우에 알림이 발생하면 어떻게 될까?

그 시간 동안 발생한 알림은 클라이언트에 도달하지 못할 것이다. 이를 방지하기 위한 것이 Last-Event-ID 헤더이다. 이 헤더는 클라이언트가 마지막으로 수신한 데이터의 id값을 의미한다. 이를 이용하여 유실된 데이터를 다시 보내줄 수 있다. 밑에서 자세히 설명할 예정이다.

서비스 (SSE) 로직

@RequiredArgsConstructor
@Service
@Slf4j
public class NotificationService {
    private static final Long DEFAULT_TIMEOUT = 1000L * 60 * 29 ;// 29분
    public static final String PREFIX_URL = "도메인 url";
    private final NotificationRepository notificationRepository;
    private final EmitterRepository emitterRepository;


    private final PlantServiceClient plantServiceClient;
    private final CircuitBreakerFactory circuitBreakerFactory;
    /**
     * SSE 연결 메서드
     * @param : MemberDto memberDto, String lastEnventId
     */
    @Transactional
    public SseEmitter subscribe(String lastEventId, String jwtToken) {
        ResponseEntity<MemberDto> joinMember = plantServiceClient.getJoinMember(jwtToken);
        Integer memberNo = joinMember.getBody().getId().intValue();
        //Emitter map에 저장하기 위한 key 생성
        String id = memberNo + "_" + System.currentTimeMillis();
        // SseEmitter map에 저장
        SseEmitter emitter = emitterRepository.save(id, new SseEmitter(DEFAULT_TIMEOUT));

        log.info("emitter add: {}", emitter);
        // emitter의 완료 또는 타임아웃 Event가 발생할 경우, 해당 emitter를 삭제
        emitter.onCompletion(() -> emitterRepository.deleteById(id));
        emitter.onTimeout(() -> emitterRepository.deleteById(id));

        // 503 에러를 방지하기 위한 더미 Event 전송
        sendToClient(emitter, id, "EventStream Created. [userId=" + memberNo + "]");

        // 클라이언트가 미수신한 Event 목록이 존재할 경우 전송하여 Event 유실을 예방
        if (!lastEventId.isEmpty()) {
            // id에 해당하는 cache 찾음
            Map<String, Object> events = emitterRepository.findAllCacheStartWithId(String.valueOf(memberNo));

            //미수신한 Event 목록 전송
            events.entrySet().stream()
                    .filter(entry -> lastEventId.compareTo(entry.getKey()) < 0)
                    .forEach(entry -> sendToClient(emitter, entry.getKey(), entry.getValue()));
        }
        return emitter;

    }
    /**
     * 알림 전송 메서드
     * 알림 받을 유저 SseEmitter 가져와서 알림 전송
     * @param : MemberDto sender, MemberDto receiver, NotifiTypeEnum type, String resource, String content
     */
    @Transactional
    public void send(MemberDto sender, MemberDto receiver, NotifiTypeEnum type, String resource, String content) {
        // 알림 생성
        Notification notification = Notification.builder()
                .senderNo(sender.getId().intValue())
                .receiverNo(receiver.getId().intValue())
                .typeEnum(type)
                .url(PREFIX_URL + type.getPath() + resource)
                .content(content)
                .isRead(false)
                .isDel(false)
                .build();
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("circuitbreaker");
        //보낸 사람 이름 찾기 위해 feignClient를 통해 호출
        ResponseEntity<MemberDto> findMember = circuitBreaker.run(() -> plantServiceClient.findById(sender.getId()),
                throwable -> ResponseEntity.ok(null));

        notification.setSenderName(findMember.getBody().getNickname());
        // SseEmitter 캐시 조회를 위해 key의 prefix 생성
        String id = String.valueOf(notification.getReceiverNo());

        //알림 저장
        notificationRepository.save(notification);

        // 로그인 한 유저의 SseEmitter 모두 가져오기
        Map<String, SseEmitter> sseEmitterMap = emitterRepository.findAllStartWithById(id);
        sseEmitterMap.forEach(
                (key, emitter) -> {
                    //캐시 저장(유실한 데이터를 처리)
                    emitterRepository.saveCache(key, notification);
                    //데아터 전송
                    sendToClient(emitter, key, NotificationResponse.toDto(notification));
                }
        );
    }
    /**
     * 클라이언트에 SSE + 알림 데이터 전송 메서드
     * @param : SseEmitter emitter, String id, Object data
     */
    private void sendToClient(SseEmitter emitter, String id, Object data) {
        try {
            log.info("event : " + data);
            emitter.send(SseEmitter.event()
                    .id(id)
                    .name("sse")
                    .data(data));
            log.info("event call: " + emitter);
        } catch (IOException ex) {
            emitterRepository.deleteById(id);
            log.error("--- SSE 연결 오류 ----", ex);
        }
    }

    /**
     * 로그인한 멤버 알림 전체 조회
     * 조회용 메서드 => (readOnly = true)
     * @param : MemberDto memberDto
     */
    public List<NotificationResponse> findAllById(Long memberNo) {
        // 채팅의 마지막 알림 조회
        List<Notification> chat
                = notificationRepository.findChatByReceiver(memberNo.intValue());

        return chat.stream()
                .map(NotificationResponse::toDto)
                .sorted(Comparator.comparing(NotificationResponse::getId).reversed())
                .collect(Collectors.toList());
    }

    /**
     * 일림 읽음 처리 메서드
     * @param : Long id
     */
    @Transactional
    public void readNotification(Long id) {
        notificationRepository.updateIsReadById(id);
    }
    /**
     * 일림 삭제 처리 메서드
     * @param : Long id
     */
    public void deleteNotification(Long[] idList) {
        for (Long id : idList) {
            Notification notification = getNotification(id);
            notificationRepository.delete(notification);
        }


    }
    //== 개별 알림 조회 ==//
    private Notification getNotification(Long id) {
        return notificationRepository.findById(id)
                .orElseThrow(ErrorCode::throwNotificationNotFound);
    }
}

subscribe()를 보면 id값을 {user_id}_{System.currentTimeMillis()} 형태로 사용하는 것을 볼 수 있습니다. 이렇게 사용하는 이유가 Last-Event-ID 헤더와 상관이 있습니다.

Last-Event-ID헤더는 클라이언트가 마지막으로 수신한 데이터의 id값을 의미한다고 했습니다. id값과 전송 데이터를 저장하고 있으면 이 값을 이용하여 유실된 데이터 전송을 다시 해줄 수 있습니다. 하지만 만약 id값을 그대로 사용한다면 어떤 문제가 있을까?

id값을 그대로 사용한다면 Last-Event-Id값이 의미가 없어집니다.

• SseEmitter emitter = emitterRepository.save(id, new SseEmitter(DEFAULT_TIMEOUT));

클라이언트의 sse연결 요청에 응답하기 위해서는 SseEmitter 객체를 만들어 반환해줘야합니다. SseEmitter 객체를 만들 때 유효 시간을 줄 수 있고 이때 주는 시간 만큼 sse 연결이 유지되고, 시간이 지나면 자동으로 클라이언트에서 재연결 요청을 보내게 됩니다.

id를 key로, SseEmitter를 value로 저장하고, SseEmitter의 시간 초과 및 네트워크 오류를 포함한 모든 이유로 비동기 요청이 정상 동작할 수 없다면 저장해둔 SseEmitter를 삭제하게 됩니다.

위에 코드는 실제로 클라이언트에 데이터를 전송하는 부분입니다. 클라이언트에 보낼 데이터 형식인 Notification 객체를 만들고, 현재 로그인 한 모든 유저의 id값을 통해 SseEmitter를 모두 가져온 이후, 데이터 캐시에도 저장해주고, 실제로 데이터 전송도 합니다.

EmitterRepository

@Repository
@Slf4j
public class EmitterRepository {
    //Map에 회원과 연결된 SSE SseEmitter 객체를 저장. 동시성 보장
    public final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    //Event를 캐시에 저장
    public final Map<String, Object> cache = new ConcurrentHashMap<>();
    //id, sseEmitter map에 저장
    public SseEmitter save(String id, SseEmitter sseEmitter) {
        emitters.put(id, sseEmitter);
        return sseEmitter;
    }
    //id와 event를 매개변수로 받아 cache 맵에 저장
    public void saveCache(String id, Object event) {
        cache.put(id, event);
    }
    //id로 시작하는 키를 가진 emitters 맵 항목을 필터링하여 맵으로 반환
    public Map<String, SseEmitter> findAllStartWithById(String id) {
        log.info("map: " + emitters);
        return emitters.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(id))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }

    //id로 시작하는 키를 가진 eventCache 맵 항목을 필터링하여 맵으로 반환
    public Map<String, Object> findAllCacheStartWithId(String id) {
        return cache.entrySet().stream()
                .filter(entry -> entry.getKey().startsWith(id))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
    }
    // id로 시작하는 키를 가진 emitters 맵 항목을 모두 제거
    public void deleteAllStartWithId(String id) {
        emitters.forEach(
                (key, emitter) -> {
                    if (key.startsWith(id)) {
                        emitters.remove(key);
                    }
                }
        );
    }
    //  emitters 맵에서 해당 id를 가진 항목을 삭제
    public void deleteById(String id) {
        emitters.remove(id);
    }


    
}

emitter 는 클라이언트가 구독을 요청하면 해당 사용자의 식별자를 키로 사용하여 맵에 저장, 이후 알림을 전송할 때 해당 사용자의 SseEmitter를 조회하기 위해 사용

eventCache는 알림을 받을 사용자의 식별자를 키로 저장하고, 해당 사용자에게 전송되지 못한 이밴트를 캐시로 저장!!

저장된 이벤트는 사용자가 구독할 때, 클라이언트로 전송되어 이벤트의 유실을 방지!

📜 emitters와 eventCache가 맵 형태인 이유?

emitters의 경우 key와 value에 각각 emitterId와 SseEmitter객체를 저장하고

eventCache는 key와 value에 각각 eventCacheId와 euentCache객체를 저장한다.

이를 통해 주어진 키를 사용하여 빠르게 데이터를 검색할 수 있고, 중복된 키를 가진 데이터를 저장할 수 없다.

concurrentHashMap은 여러 스레드에서 동시에 안전하게 데이터에 접근할 수 있는 맵 구현체이다.

알림 시스템에서는 여러 클라이언트가 동시에 구독하고 이벤트를 전송할 수 있으므로, 동시성을 제어하는 것이 중요

따라서 맵 형태를 사용함으로서 데이터 저장, 고유성 보장, 성능 등의 이점을 얻으며 효율적으로 객체를 관리할 수 있다.

2️⃣ Stomp+Redis 활용한 안읽음 기능 구현

StompHadnler 구현

  @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
        // StompCommand에 따라서 로직을 분기해서 처리하는 메서드를 호출한다.
        String username = verifyAccessToken(getAccessToken(accessor));
        log.info("StompAccessor = {}", accessor);
        handleMessage(accessor.getCommand(), accessor, username);
        return message;
    }

• ChannelInterceptor의 preSend 메서드를 먼저 Overriding
• Stomp의 이벤트에 따라서 특정 작업을 해주는 handleMessage 메서드를 작성

private void handleMessage(StompCommand stompCommand, StompHeaderAccessor accessor, String username) {
        log.info(stompCommand.toString());
        switch (stompCommand) {

            case CONNECT:
                connectToChatRoom(accessor, username);
                break;
            case SUBSCRIBE:
            case SEND:
                verifyAccessToken(getAccessToken(accessor));
                break;
        }
    }

stompCommand가 CONNECT(연결 시도)일때 connectToChatRoom 메서드가 실행되게 됩니다.

 private void connectToChatRoom(StompHeaderAccessor accessor, String username) {
        // 채팅방 번호를 가져온다.
        Integer chatRoomNo = getChatRoomNo(accessor);

        // 채팅방 입장 처리 -> Redis에 입장 내역 저장
        chatRoomService.connectChatRoom(chatRoomNo, username);
//        // 읽지 않은 채팅을 전부 읽음 처리
        chatService.updateCountAllZero(chatRoomNo, username);
        // 현재 채팅방에 접속중인 인원이 있는지 확인한다.
        boolean isConnected = chatRoomService.isConnected(chatRoomNo);

        if (isConnected) {
            chatService.updateMessage(username, chatRoomNo);
        }
    }

📕 Note

connectToChatRoom 메서드는 크게 3가지 동작을 하게 됩니다.

1. Redis에 현재 채팅방에 접속하려는 회원을 저장한다.

2. 현재 채팅방에 접속하려는 회원이 읽지 않은 채팅이 있다면, 전부 읽음 처리 해준다(readcount 0으로 업데이트)

3. 현재 채팅방에 접속중인 회원이 있다면, 채팅 리스트 다시 서버에 요청해서 받도록 처리한다.

현재 채팅방에 접속중인 회원이 있다면, 채팅 리스트를 다시 서버에 요청해서 받도록 한 이유는 아래와 같습니다.

현재 채팅방에 접속중인 회원을 A, 현재 채팅방에 접속하려는 회원을 B라고 가정하겠습니다.

A의 화면에서 B가 읽지 않은 채팅들이 안읽음 표시되게 됩니다.

B가 채팅방에 접속을 시도하면, ChannelInterceptor에 Connect 이벤트가 잡히게 되고, B가 읽지 않은 채팅들은 모두 읽음 처리 됩니다.

즉, A는 B가 채팅방에 접속하면서 채팅을 모두 읽었는데 읽지 않은 상태의 데이터를 아직 가지고 있는 것입니다.

따라서 B가 접속하면서 모든 채팅을 읽음처리 했기때문에, A에게 B가 모든 채팅을 읽었으니 최신 버전의 채팅 데이터를 서버에서 가져올 수 있도록 서버에서 알려주게 처리하였습니다.

    /**
     * 채팅방에 1명 연결됐는지 확인 메서드
     * @param : Long chatRoomNo
     */
    public boolean isConnected(Integer chatRoomNo) {
        List<ChatRoom> connectedList = chatRoomRepository.findByChatroomNo(chatRoomNo);
        return connectedList.size() == 1;
    }

- isConnected 메서드는 현재 채팅방에 한명이 접속중일때 true를 반환하도록 작성했습니다.

2️⃣ Redis를 통해 관리할 채팅방 정보

@Getter
@AllArgsConstructor
@NoArgsConstructor
@ToString
@RedisHash(value = "chatRoom")
public class ChatRoom {
    @Id
    private String id;

    @Indexed
    private Integer chatroomNo;

    @Indexed
    private String nickname;

    @Builder
    public ChatRoom(Integer chatroomNo, String nickname) {
        this.chatroomNo = chatroomNo;
        this.nickname = nickname;
    }
}

@RedisHash에 의해 레디스 chatRoom 해시 내에 저장됩니다. 각 인스턴스는 고유한 ID(id 필드)를 키로 가지며, chatroomNo와 nickname 필드는 2차 인덱스로 사용될 수 있습니다. 이를 통해 Redis에서 효율적으로 데이터를 저장하고 조회 가능합니다!

3️⃣ ChatService 수정

@Slf4j
@Service
@RequiredArgsConstructor
public class ChatService {

    private final ChatRepository chatRepository;
    private final MongoChatRepository mongoChatRepository;
    private final MessageSender sender;
    private final AggregationSender aggregationSender;
    private final MongoTemplate mongoTemplate;
    private final ChatRoomService chatRoomService;
    private final PlantServiceClient plantServiceClient;
    private final CircuitBreakerFactory circuitBreakerFactory;
    private final TokenHandler tokenHandler;

    private final NotificationService notificationService;

    /**
     * 채팅방 생성 메서드
     * 거래 게시글을 올리지 않은 사람만 호출하는 메서드
     * 구매 희망자만 채팅방을 생성 가능
     * FeignCLient를 통해 plant-service에서 거래가능 여부 확인
     * @param : MemberDto memberDto, ChatRequestDto requestDto
     */
    @Transactional
    public Chat makeChatRoom(Integer memberNo, ChatRequestDto requestDto) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("circuitbreaker");
        // 채팅을 걸려고 하는 거래글이 거래 가능 상태인지 조회해본다.
        ResponseEntity<ResponseTradeBoardDto> tradeBoardDto = circuitBreaker.run(() ->
                        plantServiceClient.boardContent(requestDto.getTradeBoardNo().longValue()),
                throwable -> ResponseEntity.ok(null));

        // 조회해온 거래글 상태가 거래완료 이라면 거래가 불가능한 상태이다.
        if (tradeBoardDto.getBody().getStatus().equals("거래완료")) {
            throw new IllegalStateException("현재 거래가능 상태가 아닙니다.");
        }
        Integer tradeBoardNo = tradeBoardDto.getBody().getId().intValue();

        //이미 해당글 기준으로 채팅을 요청한 사람과 받는 사람이 일치할 경우 체크
        if (chatRepository.existChatRoomByBuyer(tradeBoardNo, requestDto.getCreateMember(), memberNo)) {
            Chat existedChat = chatRepository.findByTradeBoardNoAndChatNo(tradeBoardNo, requestDto.getCreateMember());
            return existedChat;

        }
        if (!chatRepository.existChatRoomByBuyer(tradeBoardNo, requestDto.getCreateMember(), memberNo)) {
            Chat chat = Chat.builder()
                    .tradeBoardNo(requestDto.getTradeBoardNo())
                    .createMember(requestDto.getCreateMember())
                    .joinMember(memberNo)
                    .regDate(LocalDateTime.now())
                    .build();

            Chat savedChat = chatRepository.save(chat);


            // 채팅방 카운트 증가
            AggregationDto aggregationDto = AggregationDto
                    .builder()
                    .isIncrease("true")
                    .target(AggregationTarget.CHAT)
                    .tradeBoardNo(requestDto.getTradeBoardNo())
                    .build();

            aggregationSender.send(KafkaUtil.KAFKA_AGGREGATION, aggregationDto);
            return savedChat;
        }
        throw new IllegalArgumentException("존재하지 않는 게시글 입니다");
    }
    /**
     * 채팅방 리스트 조회 메서드
     * FeignCLient를 통해 plant-service에서 유저 정보 조회후 채팅방 만든 사람인지 확인
     * mongodb에서 채팅 메세지 보낸 시간을 내림차순으로 정렬후 첫번째 값 마지막 메세지로 세팅
     * @param : Integer memberNo, Integer tradeBoardNo
     */
    public List<ChatRoomResponseDto> getChatList(Integer memberNo, Integer tradeBoardNo) {
        List<ChatRoomResponseDto> chatRoomList = chatRepository.getChattingList(memberNo, tradeBoardNo);
        //Participant 채워야됨(username)
            chatRoomList
                    .forEach(chatRoomDto -> {
                        //param으로 넘어온 멤버가 채팅 만든 멤버일 경우 => Participant에 참가한 멤버
//                        ResponseEntity<MemberDto> byId = plantServiceClient.findById(chatRoomDto.getCreateMember().longValue());
                        if (memberNo.equals(chatRoomDto.getCreateMember())) {
                            ResponseEntity<MemberDto> memberDtoResponse = plantServiceClient.findById(chatRoomDto.getJoinMember().longValue());

                            chatRoomDto.setParticipant(new ChatRoomResponseDto.Participant(memberDtoResponse.getBody().getUsername(), memberDtoResponse.getBody().getNickname()));
                        }
                        //param으로 넘어온 멤버가 채팅방에 참가한 멤버일 경우 => Participant에 채팅방 만든 멤버
                        if (!memberNo.equals(chatRoomDto.getCreateMember())){
                            ResponseEntity<MemberDto> memberDtoResponse = plantServiceClient.findById(chatRoomDto.getCreateMember().longValue());
                            chatRoomDto.setParticipant(new ChatRoomResponseDto.Participant(memberDtoResponse.getBody().getUsername(), memberDtoResponse.getBody().getNickname()));
                        }
//                      // 채팅방별로 읽지 않은 메시지 개수를 셋팅
                        long unReadCount = countUnReadMessage(chatRoomDto.getChatNo(), memberNo);
                        chatRoomDto.setUnReadCount(unReadCount);

                        // 채팅방별로 마지막 채팅내용과 시간을 셋팅
                        Page<Chatting> chatting =
                                mongoChatRepository.findByChatRoomNoOrderBySendDateDesc(chatRoomDto.getChatNo(), PageRequest.of(0, 1));
                        if (chatting.hasContent()) {
                            Chatting chat = chatting.getContent().get(0);
                            ChatRoomResponseDto.LatestMessage latestMessage = ChatRoomResponseDto.LatestMessage.builder()
                                    .context(chat.getContent())
                                    .sendAt(chat.getSendDate())
                                    .build();
                            chatRoomDto.setLatestMessage(latestMessage);
                        }
                    });

        return chatRoomList;
    }
    /**
     * 채팅 메세지 조회 메서드
     * 채팅 메세지 조회시 해당 메세지를 읽은 것이므로 메세지 읽음 처리도 진행
     * @param : Integer chatRoomNo, Integer memberNo
     */
    public ChattingHistoryResponseDto getChattingList(Integer chatRoomNo, Integer memberNo) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("circuitbreaker");
        // member id로 조화
        ResponseEntity<MemberDto> memberDto = circuitBreaker.run(() -> plantServiceClient.findById(memberNo.longValue()),
                throwable -> ResponseEntity.ok(null));

        updateCountAllZero(chatRoomNo, memberDto.getBody().getUsername());
        List<ChatResponseDto> chattingList = mongoChatRepository.findByChatRoomNo(chatRoomNo)
                .stream()
                .map(chat -> new ChatResponseDto(chat, memberNo)
                )
                .collect(Collectors.toList());

        return ChattingHistoryResponseDto.builder()
                .chatList(chattingList)
                .email(memberDto.getBody().getEmail())
                .build();
    }
    /**
     * 메세지 전송 메서드
     * jwt 토큰에서 username 추출
     * 카프카 토픽으로 메세 전송
     * @param : Message message, String accessToken
     */
    public void sendMessage(Message message, String accessToken) {
        // member id로 조화
        ResponseEntity<MemberDto> memberDto = plantServiceClient.findByUsername(tokenHandler.getUid(accessToken));

        // 채팅방에 모든 유저가 참여중인지 확인한다.
        boolean isConnectedAll = chatRoomService.isAllConnected(message.getChatNo());
        // 1:1 채팅이므로 2명 접속시 readCount 0, 한명 접속시 1
        Integer readCount = isConnectedAll ? 0 : 1;
        // message 객체에 보낸시간, 보낸사람 memberNo, 닉네임을 셋팅해준다.
        message.setSendTimeAndSender(LocalDateTime.now(), memberDto.getBody().getId().intValue(), memberDto.getBody().getNickname(), readCount);

        // 메시지를 전송한다.
        sender.send(KafkaUtil.KAFKA_TOPIC, message);
    }
    /**
     * 알림 전송 및 메세지 저장 메서드
     * FeignCLient를 통해 plant-service에서 메세지 보낸 유저 정보 조회
     * 알림은 상대방이 읽지 않은 경우만 전송
     * @param : Message message
     */
    @Transactional
    public Message sendNotificationAndSaveMessage(Message message) {

        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("circuitbreaker");
        //메세지 저장과 알림 발송을 위해 메세지 보낸 회원 조회
        ResponseEntity<MemberDto> memberDto = circuitBreaker.run(() -> plantServiceClient.findById(message.getSenderNo().longValue()),
                throwable -> ResponseEntity.ok(null));
        // 상대방이 읽지 않은 경우에만 알림 전송
        if (message.getReadCount().equals(1)) {
            // 알람 전송을 위해 메시지를 받는 사람을 조회한다.
            Integer memberNo = chatRepository.getReceiverMember(message.getChatNo(), message.getSenderNo());
            ResponseEntity<MemberDto> receiveMember = circuitBreaker.run(() -> plantServiceClient.findById(memberNo.longValue()),
                    throwable -> ResponseEntity.ok(null));
            String content = message.getContentType().equals("image")
                                    ? "image" : message.getContent();
            // 알람을 보낼 URL을 생성한다.
            String sendUrl = getNotificationUrl(message.getTradeBoardNo(), message.getChatNo());

            //알림 전송
            notificationService.send(memberDto.getBody(), receiveMember.getBody(), NotifiTypeEnum.CHAT, sendUrl, content);
        }
        // 보낸 사람일 경우에만 메시지를 저장 -> 중복 저장 방지
        if (message.getSenderEmail().equals(memberDto.getBody().getEmail())) {
            // Message 객체를 채팅 엔티티로 변환
            Chatting chatting = message.convertEntity();
            // 채팅 내용을 저장
            Chatting savedChat = mongoChatRepository.save(chatting);
            // 저장된 고유 ID를 반환
            message.setId(savedChat.getId());
        }
        return message;
    }
    /**
     * 참가자 입장 알림 메서드
     * @param : String email, Integer chatRoomNo
     */
    public void updateMessage(String email, Integer chatRoomNo) {
        Message message = Message.builder()
                .contentType("notice")
                .chatNo(chatRoomNo)
                .content(email + " 님이 돌아오셨습니다.")
                .build();

        sender.send(KafkaUtil.KAFKA_TOPIC, message);
    }
    /**
     * 읽지 않은 메시지 채팅장 입장시 읽음 처리 메서드
     * @param : Integer chatNo, String username
     */
    public void updateCountAllZero(Integer chatNo, String username) {
        CircuitBreaker circuitBreaker = circuitBreakerFactory.create("circuitbreaker");
        ResponseEntity<MemberDto> findMember = circuitBreaker.run(() -> plantServiceClient.findByUsername(username),
                throwable -> ResponseEntity.ok(null));
        //MongoDb Update Query
        Update update = new Update().set("readCount", 0);
        //ne-> not equal
        Query query = new Query(where("chatRoomNo").is(chatNo)
                .and("senderNo").ne(findMember.getBody().getId().intValue()));

        mongoTemplate.updateMulti(query, update, Chatting.class);
    }
    /**
     * 읽지 않은 메시지 카운트 메서드
     * @param : Integer chatNo, Integer senderNo
     */
    long countUnReadMessage(Integer chatRoomNo, Integer senderNo) {
        Query query = new Query(where("chatRoomNo").is(chatRoomNo)
                .and("readCount").is(1)
                .and("senderNo").ne(senderNo));

        return mongoTemplate.count(query, Chatting.class);
    }

    private String getNotificationUrl(Integer tradeBoardNo, Integer chatNo) {
        return chatNo +
                "/" +
                tradeBoardNo;
    }


    /**
     * 판매자가 참가한  채팅방이 존재하는지 유무 처리 메서드
     * 단순 조회용 메서드라 readOnly = true
     *
     * @param : Integer tradeBoardNo,  Integer memberNo
     */
    @Transactional(readOnly = true)
    public Boolean existChatRoomBySeller(Integer tradeBoardNo, Integer memberNo) {
        return chatRepository.existChatRoomBySeller(tradeBoardNo, memberNo);
    }

    /**
     * 채팅방 삭제 메서드
     * plant-service에서 kafka를 통해 거래 게시글 삭제 요청을 받으면 삭제
     *
     * @param : Integer tradeBoardNo
     */
    @Transactional
    public void deleteChatRoom(Integer tradeBoardNo) {
        List<Integer> chatRoomNoList = chatRepository.deleteChatRoomAndReturnChatNo(tradeBoardNo);
        deleteChatting(chatRoomNoList);
    }
    @Transactional
    public void deleteChatting(List<Integer> chatRoomNoList) {
        // 중복된 chatNo 제거
        Set<Integer> uniqueChatNoSet = new HashSet<>(chatRoomNoList);

        // 중복 제거 후의 chatNo에 해당하는 채팅 데이터 삭제
        mongoTemplate.remove(query(where("chatRoomNo").in(uniqueChatNoSet)), Chatting.class);
    }
}

• ChatService에서 메시지를 DB에 저장하고 필요시 알람을 보내는 sendNotificationAndSaveMessage 메서드의 구현 부분

 // 채팅방에 모든 유저가 참여중인지 확인한다.
   boolean isConnectedAll = chatRoomService.isAllConnected(message.getChatNo());
        // 1:1 채팅이므로 2명 접속시 readCount 0, 한명 접속시 1
        Integer readCount = isConnectedAll ? 0 : 1;
        
   /**
     * 채팅방 정원 2명 찼는지 확인 메서드
     * @param : Long chatRoomNo
     */
    public boolean isAllConnected(Integer chatRoomNo) {
        List<ChatRoom> connectedList = chatRoomRepository.findByChatroomNo(chatRoomNo);
        return connectedList.size() == 2;
    }

• 이 부분을 통해 redis에 저장된 채팅방 인원을 조회하면 readCount에 값을 넣어줍니다

테스트

채팅방에 한명만 접속해있을때

gif 변환하느라 화질이 안 좋은점 양해부탁드립니다,,🙏🙏

위에 영상을 보면 한명만 접속중이기 때문에 알람이 동작하고 있는 걸 확인할 수 있습니다.

채팅방에 두명 다 접속해있을때

두명 다 접속했을땐, 알림이 가지 않는 걸 확인하실 수 있습니다!

정리

이번 포스팅까지 진행하여 채팅 마이크로 서비스 구현 과정을 모두 정리했습니다. 다음 포스팅 부터는 또 다른 마이크로 서비스인 결제 서비스 구현과정에서 포스팅 하도록 하겠습니다 🙏🙏

참고

https://develoyummer.tistory.com/112#3.2.%20Emitter%20RepositoryImpl
https://velog.io/@max9106/Spring-SSE-Server-Sent-Events%EB%A5%BC-%EC%9D%B4%EC%9A%A9%ED%95%9C-%EC%8B%A4%EC%8B%9C%EA%B0%84-%EC%95%8C%EB%A6%BC