dependencies {
	implementation 'implementation 'com.vladmihalcea:hibernate-types-52:2.17.3''
}

@Entity
@Table(name = "\"alarm\"")
@TypeDef(name="jsonb", typeClass = JsonBinaryType.class)
public class AlarmEntity {

	/*생략*/
	
	@Type(type = "jsonb") // postgres에는 json과 jsonb가 있음 - json은 일반저장, jsonb는 압축을하여 저장 및 index를 걸 수 있음.
    @Column(columnDefinition = "json")
    private AlarmArgs alarmArgs;
	
	/*생략*/
}

@TypeDef는 사용자 정의 타입에 대한 매핑을 정의한다.
위와같이 해당 엔티티 클래스레벨에 @TypeDef를 선언하고, typeClass속성에 사용할 클래스인 JsonBinaryType을, name속성에 해당 타입에 대한 이름을 지정한다.
json 타입 컬럼으로 지정할 필드에 @Type을 선언하고, type속성에 사용할 타입을 지정해준다.
type에 지정한 값은 @TypeDef에 정의한 name속성을 참조하게된다.

데이터를 조회할때는 일반적으로 데이터베이스에서 데이터를 가지고 온다.
이렇게 데이터베이스로부터 데이터를 가지고 오는 것은 부하가 많이 드는 일이다.
데이터베이스 IO(입/출력)를 사용하여 데이터베이스 로부터 데이터를 쿼링해서 가지고 와야 하는데, 이게 굉장히 비용이 많이 드는 일이다.

데이터베이스에 부하를 주지 않으면서 데이터를 조회할 수 있는 방법으로는 캐싱이 있다.

캐싱을 통해 데이터베이스 보다는 조금 더 가볍게 데이터를 저장하고 가져올 수 있다.

데이터베이스는 Query라는 질의문을 통해 질의를 함으로써 데이터를 가져오지만
일반적으로 캐싱은 Key-Value형식으로 많이 구성되어 있어, Key에 해당하는 Date를 조회를 요청하게된다.

캐싱은 여러가지 방식과, 솔루션이 있다.
그중 가장 많이 사용하는 Redis와 Local Caching에 대해 알아보자.

	Redis	Local Caching
특징	In-Memmory 데이터베이스로 Key-Value 형태의 데이터베이스. 다양한 command를 제공하며 Single Thread이다.	서버 내에 caching 하는 방법. 서버와 라이프사이클을 함께할 수 있다.
장점	여러 instance가 하나의 데이터를 공유할 수 있다. 다양한 command를 지원한다.	네트워크를 타지 않기 때문에 Redis에 비해 비교적 빠르다.
단점	Local caching에 비해서는 느리다	여러 instance로 구성된 서버의 경우 캐시를 공유할 수 없다.

• 높은 Availability를 위한 몇가지 옵션 존재

  • Sentinel

    In-Memmory : Memmory에 저장이 되기 때문에 휘발성이다.
    만약 Redis가 죽거나 했을 경우 따로 백업 옵션을 두지 않으면 데이터들이 모두 소멸되지만 캐시로 사용하는 경우 대부분 데이터베이스에 원본 데이터가 있고, 그것을 캐싱해놓는 용도기 때문에 사실상 큰 문제가 없다.
    하지만 많은 데이터를 캐싱된 채로 Redis가 죽게 된 경우 데이터베이스에서 다시 캐싱해야 하는데 바로 이때 순간적으로 데이터베이스 부하가 굉장히 높아질 것이다.
    또 Redis가 빠르기 때문에 Redis를 메인으로 쓰는 경우도 있는데, 이때 데이터 유실 발생 가능성이 있고 이 경우 Sentinel이라는 방법을 사용하여 Redis를 여러개로 구성한다.
    여러개의 Redis Instance를 하나의 Redis처럼 사용할 수 있도록 하는 방법이다.
    (클러스터링, 마스터-슬레이브 등)

  • Redis Cluster

SSE에 대한 개념 설명에 앞서, 특정 페이지를 새로고침 할 때 데이터 갱신이 일어나는 것이 아닌 회원이 특정 알림을 받아야 할 상황
즉, 게시글에 누군가가 좋아요나 댓글을 남겼을 경우, 실시간 리얼타임 까지는 아니더라도 알림을 받는방법에 대해 알아보자.

일정 주기를 가지고 서버의 API를 호출하는 방법.
(티스토리 블로그 Notification 30초 간격으로 조회 API 호출)
실시간으로 데이터가 업데이트 되지 않는다는 단점이 있다.
불필요한 요청이 발생하며 따라서 불필요한 서버 부하가 발생한다.
서비스에 유저가 적을 경우는 상관 없으나 유저가 굉장히 많은 대규모 트래픽 서비스의 경우 굉장히 많은 요청이 발생하여 서버에 큰 부하가 발생하게 된다.
다만 호환성이 좋다는 장점이 있다.

Web Browser --------- Request -------→ Server Web Browser ← Response + Data Update - Server Web Browser --------- Request -------→ Server Web Browser ← Response + Data Update - Server

서버로 요청이 들어올 경우 일정 시간동안 대기하였다가 요청한 데이터가 업데이트 된 경우 서버에서 웹브라우저로 응답을 보낸다. 즉, 일반적으로 구독과 비슷한 개념으로 API에 일정 대기시간을 걸어놓고, 특정 변경사항이 발생할 경우 해당 변경을 감지한 뒤 응답을 주도록 한다.

연결이 된 경우 실시간으로 데이터가 들어올 수 있다는 장점이 있다. 따라서 Polling보단 개선된 형태이지만 데이터 업데이트가 빈번한 경우 Polling과 유사하다.

Web Browser --------- Request -------→ Server Web Browser 　　　　　　　　　　　　　 │ Server Web Browser 　　　　　　　Data Update ↓ Server Web Browser ← Response + Data Update - Server

• 예제 코드

  • 서버: DeferredResult 클래스 사용

    • DeferredResultMamager 컴포넌트 클래스

      package com.example.longpolling;
      
      import org.springframework.stereotype.Component;
      import org.springframework.web.context.request.async.DeferredResult;
      
      import java.util.concurrent.ConcurrentHashMap;
      import java.util.concurrent.ConcurrentMap;
      
      @Component
      public class DeferredResultManager {
      
          private final ConcurrentMap<String, DeferredResult<String>> userDeferredResults = new ConcurrentHashMap<>();
      
          public void addDeferredResult(String userId, DeferredResult<String> deferredResult) {
              userDeferredResults.put(userId, deferredResult);
      
              deferredResult.onCompletion(() -> userDeferredResults.remove(userId));
              deferredResult.onTimeout(() -> userDeferredResults.remove(userId));
          }
      
          public DeferredResult<String> getDeferredResult(String userId) {
              return userDeferredResults.get(userId);
          }
      
          public void removeDeferredResult(String userId) {
              userDeferredResults.remove(userId);
          }
      }

    • Controller API

      @RestController
      public class CommentNotificationController {
      
          @Autowired
          private CommentService commentService;
      
          @PostMapping("/subscribe")
          public DeferredResult<String> subscribe(@RequestParam String userId) {
              return commentService.subscribe(userId);
          }
      
          @PostMapping("/comment")
          public String postComment(@RequestParam String comment, @RequestParam String ownerId, @RequestParam String authorId) {
              return commentService.postComment(comment, ownerId, authorId);
          }
      }

    • Service

      package com.example.longpolling;
      
      import org.springframework.beans.factory.annotation.Autowired;
      import org.springframework.stereotype.Service;
      import org.springframework.web.context.request.async.DeferredResult;
      
      @Service
      public class CommentService {
      
          @Autowired
          private CommentRepository commentRepository;
      
          @Autowired
          private DeferredResultManager deferredResultManager;
      
          public DeferredResult<String> subscribe(String userId) {
              DeferredResult<String> deferredResult = new DeferredResult<>(60000L, "No updates");
      
              deferredResultManager.addDeferredResult(userId, deferredResult);
      
              return deferredResult;
          }
      
          public String postComment(String content, String ownerId, String authorId) {
              Comment comment = new Comment();
              comment.setContent(content);
              comment.setOwnerId(ownerId);
              comment.setAuthorId(authorId);
              commentRepository.save(comment);
      
              DeferredResult<String> deferredResult = deferredResultManager.getDeferredResult(ownerId);
              if (deferredResult != null) {
                  deferredResult.setResult("새 댓글: " + content);
              }
      
              return "Comment posted";
          }
      }

  • 클라이언트

    <body>
        <h1>Long Polling Client</h1>
        <div id="messages"></div>
    
        <script>
            const userId = 'owner123'; // 댓글 주인의 고유 ID
    
            function subscribe() {
                fetch('/subscribe?userId=' + userId, {
                    method: 'POST'
                })
                    .then(response => response.text())
                    .then(message => {
                        if (message !== "No updates") {
                            document.getElementById('messages').innerText += message + "\n";
                        }
                        subscribe(); // 새 메시지를 기다리기 위해 재귀적으로 호출
                    })
                    .catch(error => {
                        console.error('Error:', error);
                        setTimeout(subscribe, 5000); // 에러가 발생하면 5초 후 다시 시도
                    });
            }
    
            document.addEventListener('DOMContentLoaded', (event) => {
                subscribe();
            });
        </script>
    </body>

서버에서 웹브라우저로 데이터를 보내줄 수 있다. 웹 브라우저에서 서버쪽으로 특정 이벤트를 구독함을 알려준다. 서버에서는 해당 이벤트가 발생하면 웹 브라우저 쪽으로 이벤트를 보내준다. 다만 서버에서 웹브라우저로만 데이터 전송이 가능하고 그 반대는 불가능하다. (웹 브라우저에서 데이터를 담아 서버로 전송하는 것은 불가능함 → 단방향) 또한 최대 동시접속 횟수가 제한되어 있다.

Web Browser -------- Subscribe ------→ Server Web Browser 　　　　　　　Data Update ↓ Server Web Browser ← Response + Data Update - Server Web Browser 　　　　　　　Data Update ↓ Server Web Browser ← Response + Data Update - Server

• 예제 코드
  • 서버 (SseEmitter 사용)

    @RestController
    public class SseController {
    
        private final static Long DEFAULT_TIMEOUT = 60L * 1000 * 60;
        private final static String ALARM_NAME = "alarm";
        private ThreadLocal<Map<String, SseEmitter>> emitterMapThreadLocal = ThreadLocal.withInitial(ConcurrentHashMap::new);
    
        @GetMapping("/subscribe")
        public SseEmitter subscribe(Authentication authentication) {
            Map<String,SseEmitter> emitterMap = emitterMapThreadLocal.get();
            SseEmitter sseEmitter = new SseEmitter(DEFAULT_TIMEOUT);
            User user = (User) authentication.getPrincipal();
            emitterMap.put(user.getId(), sseEmitter); // 수신자 id 기준 emitter 저장
            sseEmitter.onCompletion(() -> emitterMap.remove(user.getId()));
            sseEmitter.onTimeout(() -> emitterMap.remove(user.getId()));
            try {
                sseEmitter.send(
                        SseEmitter.event()
                            .id("id")
                            .name(ALARM_NAME) // EventSource의 addEventListner로 등록한 이벤트명과 일치하는 이벤트명을 입력해준다.
                            .data("connect completed")
               );
            } catch (IOException e) {
                throw new SnsApplicationException(ErrorCode.ALARM_CONNECT_ERROR);
            }
            return sseEmitter;
        }
    
        @PostMapping("/send")
        public void sendNotification(Integer alarmId, Integer userId) {
            Map<String,SseEmitter> emitterMap = emitterMapThreadLocal.get();
            SseEmitter sseEmitter = emitterMap.get(userId); // 수신자 id 기준 emitter 조회
            try {
                sseEmitter.send(
                        SseEmitter.event()
                        .id(alarmId.toString()) // 반환받을 클라이언트의 EventSource에서 Data를 구분/식별할 수 있는 PK와 같은 값...
                        .name(ALARM_NAME) // EventSource의 addEventListner로 등록한 이벤트명과 일치하는 이벤트명을 입력해준다.
                        .data("new alarm")
                );
            } catch (IOException e) {
                emitterMap.remove(user.getId());
                throw new RuntimeException();
            }
        }
    }

    • JwtTokenFilter (Header값 추출...)

      RequiredArgsConstructor
      public class JwtTokenFilter extends OncePerRequestFilter {
      private final String key;
      private final UserService userService;
      
          private final static List<String> TOKEN_IN_PARAM_URLS = List.of("/api/v1/users/alarm/subscribe");
          
          @Override
          protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException {
              
              try {
                  final String token;
                  if (TOKEN_IN_PARAM_URLS.contains(request.getRequestURI())) {
                      log.info("Request with {} check the query param", request.getRequestURL());
                      token = request.getQueryString().split("=")[1].trim();
                  } else {
                      // get Header
                      final String header = request.getHeader(HttpHeaders.AUTHORIZATION);
                      if (header == null || !header.startsWith("Bearer ")) {
                          log.error("Error occurs while getting header. header is null or invalid {}", request.getRequestURL());
                          filterChain.doFilter(request, response);
                          return;
                      }
                      token = header.split(" ")[1].trim();
                  }
      
                  /* 토큰 유효 검증 */
                  if (JwtTokenUtils.isExpired(token, key)) {
                      log.error("Key is expired");
                      filterChain.doFilter(request, response);
                      return;
                  }
      
                  /* 토큰으로 부터 username 획득 */
                  String userName = JwtTokenUtils.getUserName(token, key);
      
                  /* username을 통해 회원 엔티티 조회 */
                  log.info("jwtTokenFilter - loadByUsername 호출");
                  User user = userService.loadByUsername(userName);
                  log.info("jwtTokenFilter - loadByUsername 호출 종료");
      
                  /* 회원 정보와 Role 각각 Security의 Pirncipal과 Authorities에 저장 */
                  UsernamePasswordAuthenticationToken authenticationToken =
                          new UsernamePasswordAuthenticationToken( //princlipal, credential, authorities
                          user, null, user.getAuthorities()
      //                    user, null, List.of(new SimpleGrantedAuthority(user.getUserRole().toString()))
                  );
      
                  /* Detail 초기화 - request 정보를 함께 넣어준다. */
                  authenticationToken.setDetails(new WebAuthenticationDetailsSource().buildDetails(request));
                  SecurityContextHolder.getContext().setAuthentication(authenticationToken);
      
              } catch (RuntimeException e) {
                  log.error("Error occurs while validating. {}", e.toString());
                  filterChain.doFilter(request, response);
                  return;
              }
              filterChain.doFilter(request, response); // 종료 후 다음 Filter로 request와 response를 넘겨준다.
          }
      }    
          

  • 클라이언트 (EventSource 사용)

    const eventSource = new EventSource("http://localhost/subscribe?token=" + localStorage.getItem('token'));
    
    eventSource.addEventListener("open", function (event) {
    console.log("connection opened");
    });
    
    eventSource.addEventListener("alarm", function (event) {
    console.log(event.data);
    handleGetAlarm();
    });
    
    eventSource.addEventListener("error", function (event) {
        console.log(event.target.readyState);
        if (event.target.readyState === EventSource.CLOSED) {
        console.log("eventsource closed (" + event.target.readyState + ")");
        }
        eventSource.close();
    });

서버와 웹브라우저 사이에 양방향 통신이 가능한 방법이다.
채팅과 같은 프로그램에서 많이 사용하는 방법이다.

https://kafka.apache.org/downloads

위 다운로드 사이트에서 Binary 파일 다운로드 후 띄어쓰기가 없는 경로에 압축 해제( C:\Kafka )

• Command Line 명령어를 통한 Zookeeper/Kafka 기동 및 Topic생성, consumer출력
  1. Zookepper 서버 기동
     C:\Kafka> .\bin\windows\zookeeper-server-start.bat config\zookeeper.properties
  2. Kafka 서버 기동
     C:\Kafka>.\bin\windows\kafka-server-start.bat config\server.properties
  3. Topic 생성
     C:\Kafka>.\bin\windows\kafka-topics.bat --create --topic alarm --bootstrap-server localhost:9092
  4. consumer 내역 리얼타임 출력
     C:\Kafka>.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic alarm --from-beginning
  5. Topic 목록 확인
     C:\Kafka>.\bin\windows\kafka-topics.bat --bootstrap-server localhost:9092 --list

카프카 공식 문서

카프카는 간단하게 메시징 큐라고 말할 수 있다.

메시지를 생성하는 Producer, 메시지(이벤트)를 받아서 처리하는 Consumer가 있다.
이때 Producer가 생성한 메시지(이벤트)가 어딘가에 저장이 되어야 Consumer가 메시지(이벤트)를 읽어올 수 있다.
Producer가 생성한 메시지(이벤트)는 중간에서 다리 같은 역할을 하는 Broker 에 파일로 작성(저장) 된다.
Consumer 는 Broker 에 저장된 메시지를 읽어온다.
Producer 의 경우 메시지(이벤트)를 생성할 때 Key라는 값을 설정해야 한다.
Key는 메시지(이벤트)가 어떻게 파티션이 될지를 정하는 값이다.

Producer 가 메시지를 생성하고 Broker 저장할 때 Broker 내부에 지정할 수 있는 영역과 같다.
이는 Topic을 중심으로 메시지가 생성이 되고, Topic으로 부터 메시지를 소비하는 구조라고 생각하면 된다.
이를 간단하게 말하자면 Producer 와 Consumer 를 이어주는 Key같은 값이다.

이처럼 카프카에는 여러가지의 Topic이 생성 될 수가 있다.
Producer 는 Broker 내부의 Topic에 메시지(이벤트)를 전달하고, Consumer는 해당 Topic에서 메시지를 읽어오는 방식으로 동작하게 된다.

이렇게 전달된 메시지(이벤트)는 파티션닝 되어 저장되게 된다.

예를들어 메시지가 단순하게 파티셔닝이 된다면, 게시글을 업데이트를 했다고 가정한다. (해당 포스트에 업데이트된 정보가 메시지(이벤트)로 Topic에 전달된다.)

1. 게시글 제목 A를 B로 바꿨다는 이벤트
2. 게시글 제목 B를 C로 바꿨다는 이벤트

결론적으로는 A를 C로 바꾼것, 최종적인 데이터는 C 이긴 하지만, 이것을 Consumer에 의해 읽어들여 B로 변경 후 C로 변경했다 라는 순서를 보장하기 위해서 가장 먼저 읽어야 될 것은 B로 변경된 것을 읽어야 하고, 그 다음 C로 변경된 것을 읽어야 한다.
만약 순서가 바뀌어서 C로 변경된 것을 먼저 읽게 되면 C로 변경 후 B로 변경된 것 처럼 보일 수 있다.
이 순서가 보장이 되기 위해서는 Key값이 같아야 한다.
Key값을 바탕으로 파티셔닝이 되는데 Consumer 가 읽을 때 파티션을 차례대로 읽는다.
이는 같은 파티션에서는 나중에 생성된 데이터가 뒤에 오기 때문에 컨슈머에서 더 늦게 처리되는 것이 보장이 되지만, 파티션이 다른 경우에 컨슈머가 파티션을 읽었을 때 더 늦게 저장된 이벤트를 더 빨리 읽게 되면 해당 파티션에 있는 메시지가 먼저 처리될 수 있다.

Partition0에 B로 변경한 이벤트가 저장되었고, Partition1에 C로 변경한 이벤트가 저장되었다고 가정한다.

Consumer 는 Consumer Group이라는 것이 존재하며, 해당 Group에는 여러 Consumer 들이 존재한다.
이는 같은 토픽을 나눠 읽기 위함으로 설계된것이다.

Consumer1이 Partition0에 접근하고 Consumer2가 Partition1에 접근한다.
이때 특정 상황에 의해 아주 불행하게도 Consumer2가 조금 더 빠르고 Consumer1이 살짝 느려져 버렸다.
이에 Consumer2가 먼저 partition1의 C로 변경된 메시지를 처리해 버리고 Consumer1이 다음 메시지인 B로 변경된 메시지를 처리하게 되면 결론적으로 게시글을 B로 바꾼 뒤 C로 바꾼것인데 글이 B로 변경된 것 처럼 보일것이다.

이를 방지하기 위해서 Key값을 설정하는 것이다.
Key값을 기준으로 파티션이 나눠지기 때문에 같은 Key를 가지고 있다면 같은 파티션에 메시지가 적재되게 돼서 하나의 컨슈머가 해당 파티션을 처리되어 절대적으로 메시지의 순서가 바뀌는 현상이 방지된다.

이후 Consumer에서 메시지를 읽은 뒤 완료 신호로 Ack를 날리게 된다.
(Ack가 전송되야 비로서 메시지가 처리되었다고 표시되는것)

• 단위별로 10개를 읽으면 날리는 배치성 Ack
• 1개 읽을때 마다 날리는 Ack
• Command에 의해 Ack시점을 결정하는 방법