SpringBoot 설정

1. build.gradle에 웹소켓과 관련 STOMP 의존성 추가

implementation 'org.springframework.boot:spring-boot-starter-websocket'
implementation 'org.webjars:sockjs-client:1.5.1'
implementation 'org.webjars:stomp-websocket:2.3.4'

2. WebSocketConfig 추가

1) WebSocketMessageBrokerConfigurer을 상속 받은 Configuration 클래스 생성

2) 클래스에 @EnableWebSocketMessageBroker 어노테이션 추가: Spring WebSocket 메시징 시스템에서 브로커를 활성화

3) registerStompEndpoints override:

• registry.addEndpoint()에 웹소켓 연결을 실행할 수 있는 엔드포인트 지정
• setAllowedOrigins()으로 CORS 에러가 발생하지 않도록 origin 허용
• withSockJS()로 SockJS 사용 여부 설정(사용하지 않을 거라면 withSockJS()를 생략하면 된다.)

4) configureMessageBroker override:

• setApplicationDestinationPrefixes: 클라이언트로부터 메시지를 받을 엔드포인트의 접두사 설정

• enableSimpleBroker: 메시지를 클라이언트가 구독한 경로로 전달하기 위한 메시지 브로커 설정 (참고로 내 경우에는 서버에서 특정 로직 수행 후 클라이언트에 메시지를 전송하는 기능만 구현했기 때문에 configureMessageBroker는 실제 필요하지 않았다. 다만 참고를 위해 적어두었다. MessageBroker 기능을 사용하기 위해서는 @MessageMapping @SendTo 어노테이션을 이용해 메시지를 수신하고 전달하게 된다.)

  @Configuration
  @EnableWebSocketMessageBroker
  public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
  
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry
                // 클라이언트가 서버로 메시지를 보낼 때 엔드포인트의 접두사
                .setApplicationDestinationPrefixes("/pub")
                // 클라이언트로 구독한 경로로 메시지를 전달하기 위해 사용하는 내장 메시지 브로커 활성화
                .enableSimpleBroker("/sub");
    }
  
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOrigins("http://localhost:3000")
                .withSockJS();
    }
  }
  

<br>

### 3. WebSocket 메시지 전송
위에서 언급한 것처럼 웹소켓 메시지 브로커 기능을 이용하여 웹소켓으로 메시지를 받고, 그 메시지를 클라이언트에 전달하기 위해서는 컨트롤러에 @MessageMapping, @SendTo 어노테이션을 붙여서 기능을 구현하면 된다. 

내 경우에는 특정 API가 호출되면 로직을 처리한 후 대시보드에 내용을 실시간으로 업데이트하기 위해서, 즉 **메시지를 전달하는 기능**만 필요해서 SimpMessagingTemplate를 사용하였다. SimpMessagingTemplate의 convertAndSend 메서드에 메시지를 전달할 엔드포인트와 payload를 적어주면 된다.

@Service @RequiredArgsConstructor public class DashBoardService {

private final SimpMessagingTemplate simpMessagingTemplate;

public void processOrder( ) {
    simpMessagingTemplate.convertAndSend("/dash-board", "TEST 메시지");
}

}

<br>
<hr>

# React 설정

먼저 리액트 STOMP 관련 라이브러리를 찾아봤을 때, 크게 2가지 라이브러리를 사용하는 것 같았다.
stompjs와 @stomp/stompjs가 그것인데, stompjs가 좀 더 설정이 간단하지만 고급 설정에 제한이 있다는 단점이 있었고 @stomp/stompjs는 설정은 조금 더 복잡하지만 Heartbeat, 자동 재연결 같은 고급 기능을 제공한다는 장점이 있었다. 
나는 일단 두 가지 다 구현해 보았다.

### 1. stompjs 사용
1) 라이브러리 설치(sockjs도 사용하려고 sockjs-client도 함께 설치)

npm install stompjs sockjs-client

2) 웹소켓 연결 코드 
- 해당 페이지 렌더링 시 웹소켓을 연결하기 위해 useEffect 안에 웹소켓 연결 로직 작성

useEffect(() => {
  const socket = new SockJS('http://localhost:9090/ws'); // 서버 url + 서버에서 설정한 웹소켓 연결 엔드포인트
  const stompClient = Stomp.over(socket);

  stompClient.connect({}, () => {
      stompClient.subscribe('/dash-board', (message) => {
          console.log(message.body); 
      });
  });

  return () => {
      stompClient.disconnect(); 
  };

}, []);

<br>

### 2. @stomp/stompjs 사용
1) 라이브러리 설치

npm install @stomp/stompjs

<br>
2) 웹소켓 연결 코드 

**SockJS**를 사용하는 경우에는 **webSocketFactory**를 설정해주고, 사용하지 않는 경우에는 **brokerURL**만 적어주면 된다. 
참고로 둘 다 적어준 경우에는 webSocketFactory가 우선 적용(?)되는 것 같았다. (= 서버에 withSockJS() 설정이 있어야 웹소켓이 연결됨)

const [client, setClient] = useState(null);

useEffect(() => { const connect = () => { const clientdata = new Client({ // webSocketFactory: () => new SockJS("http://localhost:9090/ws"), // 서버에 withSockJS() 설정이 있는 경우 brokerURL: "ws://localhost:9090/ws", // 서버에 withSockJS() 설정이 없는 경우 debug: function (str) { console.log(str); }, reconnectDelay: 9000, // 재연결 시간 heartbeatIncoming: 4000, heartbeatOutgoing: 4000, });

  clientdata.onConnect = () => {
    clientdata.subscribe("/dash-board", (message) => {
      console.log("Received message:", message.body);
    });
  };

  clientdata.onStompError = (frame) => {
    console.error("STOMP error:", frame.headers["message"]);
  };

  clientdata.activate();
  setClient(clientdata);
};

connect();

return () => {
  if (client) {
    client.deactivate();
  }
};

}, []);

<br>
<hr>
<br>
이상 내가 기억하려고 적어본 웹소켓 서버/클라이언트 설정 방법이다. 나중에 기회가 되면 채팅 어플리케이션도 만들어 보고 싶다! 😊

[백준] 1919 - 애너그램 만들기

애너그램은 결국 순서에 상관없이 두 단어의 철자의 알파벳 개수가 같으면 조건을 만족하기 때문에, 문제를 풀기 위해서는 두 단어의 알파벳 개수 차이를 전부 더해주면 된다.

그러기 위해서 우선 행의 길이 2(두 개의 단어), 열의 길이 26(알파벳)을 가진 2차원 배열을 선언해주고, 각 단어의 철자를 하나씩 살펴보며 해당하는 알파벳의 개수를 증가시킨다.

마지막에는 알파벳 순서대로 반복문을 돌면서 0번째 배열과 1번째 배열의 알파벳 개수 차이를 delete에 합산한 후 출력해주면 된다.

아래는 전체 소스코드이다.

// 1919 - 애너그램 만들기
public class Main {

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));

        int[][] alphabet = new int[2][26];
        for (int i = 0; i < 2; i++) {
            String input = br.readLine();
            for (int j = 0; j < input.length(); j++) {
                alphabet[i][input.charAt(j) - 'a']++;
            }
        }
        int delete = 0;
        for (int j = 0; j < alphabet[0].length; j++) {
            delete += Math.abs(alphabet[0][j] - alphabet[1][j]);
        }

        bw.write(delete+"");
        bw.close();
    }
}

[백준] 1181 - 단어 정렬

이것도 1431-시리얼 번호와 마찬가지로 Comparator로 정렬 기준을 만들어서 풀이하였다. 다만 중복된 단어는 하나만 남기고 제거해야한다는 조건이 있어서 정렬 + 중복제거가 가능한 SortedSet 자료구조를 사용하였다.

아래는 전체 소스코드이다.

// 1181 - 단어 정렬
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        SortedSet<String> set = new TreeSet<>((a, b) -> {
            if(a.length() != b.length()) return a.length() - b.length();
            return a.compareTo(b);
        });
        int N = Integer.parseInt(br.readLine());
        for (int i = 0; i < N; i++) {
            set.add(br.readLine());
        }

        for (String word : set) {
            bw.write(word);
            bw.write("\n");
        }

        br.close();
        bw.close();
    }
}

[백준] 1431 - 시리얼 번호

사실 이 문제는 많이 까다롭거나 어려운 문제는 아니었는데 내 실수로 헤맸었던 문제이다. 문제 자체는 문자열을 정렬하기 위한 comparator를 구현하면 되는 것이었는데, 내가 정렬 기준을 세울 때 잘못 생각한 부분이 있었다.

먼저 문제의 요구 조건 1.A와 B의 길이가 다르면, 짧은 것이 먼저 온다. 2.만약 서로 길이가 같다면, A의 모든 자리수의 합과 B의 모든 자리수의 합을 비교해서 작은 합을 가지는 것이 먼저온다. (숫자인 것만 더한다) 3. 만약 1,2번 둘 조건으로도 비교할 수 없으면, 사전순으로 비교한다. 숫자가 알파벳보다 사전순으로 작다. 중에 2번이 문제였는데... 나는 각 자리수의 '합'을 구해서 비교하는 것이므로 실제값이 아닌 아스키 문자 값으로 더해도 합의 크기를 비교하는 데에는 문제가 없을 것이라고 생각했다. 그래서 아래와 같이 구현했고 결과는...

실제로 테스트케이스는 모두 통과하는데 자꾸 제출하고 나면 거의 바로 틀렸습니다가 나오더라..^-ㅜ 결국 챗GPT의 도움을 받아서 반례 케이스를 얻을 수 있었다. 아스키 값의 합은 같지만 실제 숫자 값의 합은 다른 경우이다.

문자열 "39":
아스키 값 합: 51 + 57 = 108
숫자 값 합: 3 + 9 = 12

문자열 "48":
아스키 값 합: 52 + 56 = 108
숫자 값 합: 4 + 8 = 12

그래서 다시 아래처럼 -'0' 처리를 해줘서 실제 숫자값을 합하도록 수정했더니 통과했다.

그리고 아래는 전체 소스코드인데, 사전순으로 정렬하는 부분은 return a.compareTo(b); 로 String의 compareTo 메서드를 이용해도 동일한 결과가 나온다.

import java.io.*;
import java.util.Arrays;

// 1431 - 시리얼 번호
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        int N = Integer.parseInt(br.readLine());
        String[] arr = new String[N];
        for (int i = 0; i < N; i++) {
            arr[i] = br.readLine();
        }
        br.close();
        Arrays.sort(arr, (a, b) -> {
            if (a.length() != b.length()) {
                return a.length() - b.length();
            }

            int sumA = 0;
            int sumB = 0;
            for (int i = 0; i < a.length(); i++) {
                if(a.charAt(i) >= '0' && a.charAt(i) <= '9'){
                    sumA += a.charAt(i) - '0';
                }
            }
            for (int i = 0; i < b.length(); i++) {
                if(b.charAt(i) >= '0' && b.charAt(i) <= '9'){
                    sumB += b.charAt(i) - '0';
                }
            }
            if(sumA != sumB) return sumA - sumB;

            // 사전순 정렬
            for (int i = 0; i < a.length(); i++) {
                if(a.charAt(i) != b.charAt(i)) {
                    return a.charAt(i) - b.charAt(i);
                }
            }

            return 0;

        });

        for (int i = 0; i < arr.length; i++) {
            bw.write(arr[i]);
            bw.write("\n");
        }
        bw.close();
    }
}

[백준] 6443 - 팰린드롬 만들기

처음에는 문제를 읽고 '음.. 순열 문제군... 근데 중복되는 건 어떻게 제거하지?' 하다가 SortedSet에 넣어보기로 했었다. 결과는 역시나 예상대로 메모리 초과! 그래서 그 다음에는 순열을 돌 때 이전 인덱스와 문자가 같으면... 어떻게 어떻게 지지고 볶고... 해서 중복을 제거하면 되겠지!? 했는데, 중복되는 문자가 2개일 때는 먹히는데 늘어나면 안 먹히더라 ^_ㅜ 그렇다고 검사하는 범위가 늘어나면 또 메모리나 시간초과 날 것 같고...

결국 다른 사람들 풀이를 보고 힌트를 얻었는데, 문자 그 자체를 반복/조합해서 순열로 만드는 게 아니라 문자의 개수를 이용해서 중복없는 순열을 만드는 방법이 있었다. 심지어는 visited 배열도 불필요!

아래 Perm 메서드가 그걸 구현한 것인데, 알파벳의 개수를 기록한 int 배열 alphabet, 결과 문자를 담을 배열인 result, 현재 뽑은 문자를 카운트할 r, 뽑아야할 개수인 count(사실 result.length를 사용해도 된다.)를 인자로 사용한다. 그렇게 앞에서부터 alphabet배열을 돌면서 사용한 문자는 개수에서 빼주고, 재귀함수를 호출한 후 재귀함수에서 빠져나오면 다시 개수를 더해준다.

아래는 전체 소스코드이다.

import java.io.*;

// 6443 - 애나그램
public class Main {
    static BufferedWriter bw;

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
         bw = new BufferedWriter(new OutputStreamWriter(System.out));

        int N = Integer.parseInt(br.readLine());
        for (int i = 0; i < N; i++) {
            int[] alphabet = new int[26];
            String input = br.readLine();

            // 알파벳 개수 카운트
            for (int j = 0; j < input.length(); j++) {
                alphabet[input.charAt(j) - 'a']++;
            }

            char[] result = new char[input.length()];
            perm(alphabet, result, 0, input.length());

        }
        bw.close();
    }

    static void perm(int[] alphabet, char[] result, int r, int count) throws IOException {
        if (r == count) {
            bw.write(String.valueOf(result));
            bw.write("\n");
        }
        for (int i = 0; i < alphabet.length; i++) {
            if (alphabet[i] > 0) {
                alphabet[i]--;
                result[r] = (char)( i+'a');
                perm(alphabet, result, r+1, count);
                alphabet[i]++;
            }
        }
    }
}

[백준] 1254 - 팰린드롬 만들기

규칙을 찾으면 코드 구현 자체는 그렇게 어렵지 않는 문제였다. (물론 늘 그렇지만 규칙을 찾는 게 어렵다는 게 문제...)

내가 생각한 규칙은 문자열의 맨 뒤는 고정하고, 앞에서부터 차례로 문자열을 확인하여 현재의 문자열이 팰린드롬일 때까지 인덱스를 1씩 늘려가는 것이었다. 그리고 현재 문자열이 팰린드롬이라면, 현재 인덱스의 값만큼 문자열 길이에 추가해주면 된다. (= 전체 문자열 뒤에 현재 인덱스까지의 문자열을 추가해주는 것)

아래는 전체 소스코드이다.

// 1254 - 팰린드롬 만들기
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));

        String input = br.readLine();
        br.close();
        int i = 0;
        while (i < input.length()) {
            if(isPalindrome(input.substring(i))) break;
            i++;
        }
        bw.write(String.valueOf(input.length() + i));
        bw.flush();
        bw.close();
    }

    static boolean isPalindrome(String string) {
        boolean result = true;
        for (int i = 0; i < string.length() / 2; i++) {
            if (string.charAt(i) != string.charAt(string.length() - 1 - i)) {
                result = false;
                break;
            }
        }
        return result;
    }
}

[백준] 10942 - 팰린드롬?

사실 처음에는 주어진 질문 M번에 대해 일일이 팰린드롬인지 체크를 해주는 방식으로 풀었는데 그래도 통과는 할 수 있었다. 근데 풀면서도 이게 맞나?? 값을 저장해놓고 쓸 수 있을 것 같은데... DP??? 라는 생각이 들어서 다시 시도 해보았다.

역시 DP 점화식 세우기가 어려워서 애먹었지만 결과적으로 시간을 반 정도로 줄일 수 있었다. 나중의 나를 위해... 내가 푼 방법을 차례로 적어보겠다.

1. 우선 칠판에 적힌 수를 입력받을 int 배열 nums와 isPalindrome이라는 2차원 boolean 배열을 선언한다. isPalindrome의 행 인덱스를 문자열의 시작, 열 인덱스를 문자열의 끝으로 간주한다.

2. 칠판에 적힌 숫자를 입력받을 때 문자 하나는 무조건 팰린드롬이므로 시작 인덱스와 끝 인덱스가 같은 경우는 isPalindrome 값을 true로 넣어준다. 또한 현재 주어진 문자와 nums[현재 인덱스-1]의 문자와 동일한 경우에도 길이가 2인 팰린드롬이므로 isPalindrome 값을 true로 넣어준다.

3. 길이가 1, 2인 경우는 체크했으므로 이제 길이가 3(간격이 2)인 경우부터 길이가 N (간격이 N-1)인 경우까지 차례로 체크해주면 된다. 아래 이중 포문에서 i는 간격을 하나씩 늘려가는 반복문이고, j는 시작인덱스이다. 이렇게 for문을 돌면서 현재 체크하려는 부분의 양쪽 끝을 제외한 문자열이 팰린드롬인지 확인하고, 이것이 팰린드롬이라면 양쪽 끝의 문자가 서로 같은 값인지까지 체크해주면 된다.

아래가 전체 소스코드이다.

// 10942 - 팰린드롬?
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        StringTokenizer st;
        int N = Integer.parseInt(br.readLine());

        int[] nums = new int[N+1];
        // 행 인덱스는 문자열의 시작, 열 인덱스는 문자열의 끝
        boolean[][] isPalindrome = new boolean[N + 1][N + 1];
        st = new StringTokenizer(br.readLine());
        for (int i = 1; i < N+1; i++) {
            nums[i] = Integer.parseInt(st.nextToken());
            // 문자 하나는 무조건 팰린드롬
            isPalindrome[i][i] = true;
            // 이전 인덱스의 문자와 현재 문자가 동일한 경우에도 팰런드롬
            if(nums[i-1] == nums[i]) isPalindrome[i-1][i] = true;
        }
        for (int i = 2; i < N; i++) { // i는 간격
            for (int j = 1; j + i < N + 1; j++) { // i+j가 마지막 인덱스보다 작거나 같을 때까지만 반복 
                isPalindrome[j][j+i] = isPalindrome[j+1][j+i-1] && nums[j] == nums[j+i]; // 양쪽 끝을 제외한 문자열이 팰린드롬인지 && 양쪽 끝이 동일한 문자인지 체크
            }
        }
        int M = Integer.parseInt(br.readLine());
        for (int i = 0; i < M; i++) {
            st = new StringTokenizer(br.readLine());
            int start = Integer.parseInt(st.nextToken());
            int end = Integer.parseInt(st.nextToken());
            bw.write(isPalindrome[start][end] ? "1\n" : "0\n");
        }

        br.close();
        bw.flush();
        bw.close();
    }

// 처음에 사용했던 팰린드롬 체크 메서드
    static boolean checkItIsPalindrome(int start, int end, int[] nums) {
        boolean isPalindrome = true;
        for (int i = 0; i < (end-start+1)/2; i++) {
            if (nums[start+i] != nums[end - i]) {
                isPalindrome = false;
                break;
            }
        }
        return isPalindrome;
    }
}

[백준] 9996 - 한국이 그리울 땐 서버에 접속하지

처음엔 뭐야 쉬운 문제네~ 하고 풀었는데 자꾸 60% 대에서 틀렸습니다가 나와서 결국 반례 푸는 법을 찾아봤던 문제. 어떤 반례가 있냐하면, 예를 들어 주어진 패턴이 'ab*bc' 이고, 주어진 파일 이름이 'abc'인 경우에 이 파일의 이름은 ab로 시작하는 것도 맞고 bc로 끝나는 것도 맞지만 'ab*bc' 패턴과는 일치하지 않는다.

그래서 단순히 패턴의 시작과 끝 문자열과 파일 이름의 앞뒤 문자열을 비교만 하면 안되고 파일 이름의 길이가 *을 제외한 패턴 문자열의 길이와 같거나 큰지도 체크해줘야한다. 다음 코드에서 'patternString.length()-1 <= input.length()'에 해당하는 부분이 그것이다.

// 9996 - 한국이 그리울 땐 서버에 접속하지
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));

        int N = Integer.parseInt(br.readLine());
        String patternString = br.readLine();
        String[] pattern = patternString.split("\\*");
        for (int i = 0; i < N; i++) {
            String input = br.readLine();
            int i1 = input.indexOf(pattern[0]);
            int i2 = input.lastIndexOf(pattern[1]);

            bw.write(patternString.length()-1 <= input.length() && i1 == 0 && i2 == input.length() - pattern[1].length() ? "DA" : "NE");
            bw.write("\n");
        }
        br.close();
        bw.flush();
        bw.close();
    }
}

백준 1167 - 트리의 지름

이 문제는 아무리 생각해도 어떻게 풀어야할지 감이 안 와서 검색해보니, 트리의 성질과 관련한 공식에 대한 이해가 필요한 문제였다.

루트에서 가장 먼 정점 A를 구하고, A에서 가장 먼 정점 B를 구하면(참고로 B가 루트 노드일 수도 있다) A와 B의 거리가 트리의 지름이라는 것이었다.

이걸 처음 봤을 때는 오... 그렇구나.. 하면서도 수학적으로 증명이 된 건지 궁금했는데 다행히 증명 과정을 잘 적어놓은 블로그들도 많이 있었다. 물론...(^^) 증명들을 읽는다고 바로 이해가 가지는 않았는데, 개인적으로는 아래 블로그 설명이 가장 명확히 납득시켜 주었던 것 같다. https://johoonday.tistory.com/217

이제 증명에 대한 이해까지 끝냈으면 문제를 풀 차례! 공식을 생각해내는 게 어려워서 그렇지 풀이 자체는 어렵지 않았다. dfs(혹은 bfs)를 2번 수행해서 루트에서 가장 먼 노드와 그 노드에서 가장 먼 노드와의 거리를 구하면 된다.

아래는 전체 소스코드이다.

// 1167 - 트리의 지름
public class Main {
    static List<List<int[]>> list; // 노드의 번호와 간선의 거리 둘 다 저장해야하므로 int[] 

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        StringTokenizer st;
        int N = Integer.parseInt(br.readLine());
        list = new ArrayList<>();
        for (int i = 0; i < N + 1; i++) {
            list.add(new ArrayList<>());
        }

        for (int i = 0; i < N; i++) {
            st = new StringTokenizer(br.readLine());
            int node = Integer.parseInt(st.nextToken());
            int n;
            int distance;
            while (true) {
                n = Integer.parseInt(st.nextToken());
                if (n == -1) {
                    break;
                }
                distance = Integer.parseInt(st.nextToken());
                list.get(node).add(new int[]{n, distance});
            }
        }
        br.close();

        boolean[] visited = new boolean[list.size()];
        visited[1] = true;
        // 루트 노드에 대한 dfs를 수행하고 가장 먼 거리와 노드 번호를 구한다.
        int[] fromRoot = dfs(1, visited, 0);
        visited = new boolean[list.size()];
        visited[fromRoot[1]] = true;
        // 가장 먼 노드에서부터 다시 dfs를 수행
        int[] farthest = dfs(fromRoot[1], visited, 0);

        bw.write(farthest[0] + "");
        bw.flush();
        bw.close();
    }

    static int[] dfs(int index, boolean[] visited, int distance) {
        int maxDistance = distance;
        int farthestNode = index;

        for (int[] node : list.get(index)) {
            if (!visited[node[0]]) {
                visited[node[0]] = true;
                int[] disAndNode = dfs(node[0], visited, distance + node[1]);
                if (disAndNode[0] > maxDistance) {
                    maxDistance = disAndNode[0];
                    farthestNode = disAndNode[1];
                }
            }
        }
        return new int[]{maxDistance, farthestNode};
    }
}

처음엔 트리!!??? 라고 해서 괜히 겁먹었지만 문제를 천천히 읽어보니 결국은 dfs, bfs 문제였다. dfs, bfs 둘 다 비슷하게 메모리와 시간이 소요된다.

간단한 bfs, dsf 문제이기 때문에 다른 풀이들과 전반적인 코드는 비슷하다. 다만 visited 배열과 부모 노드를 담는 배열을 따로 따로 선언할 필요가 없어 보이길래 그냥 visited 배열을 int 배열로 만들어서 값이 0인 경우에는 방문하지 않은 노드로 체크하고, 해당 인덱스의 배열 값에 부모 노드의 숫자를 넣도록 코드를 작성했다.

import java.io.*;
import java.util.*;

// 11725 - 트리의 부모 찾기
public class Main {
    static List<List<Integer>> list;

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        StringTokenizer st;
        int N = Integer.parseInt(br.readLine());
        list = new ArrayList<>();
        for (int i = 0; i <= N; i++) {
            list.add(new ArrayList<>());
        }
        for (int i = 0; i < N-1; i++) {
            st = new StringTokenizer(br.readLine());
            int n1 = Integer.parseInt(st.nextToken());
            int n2 = Integer.parseInt(st.nextToken());
            list.get(n1).add(n2);
            list.get(n2).add(n1);
        }
        int[] visited = new int[list.size()];
        visited[1] = 1;
        bfs(1, visited);
//        dfs(1, visited);

        for (int i = 2; i < visited.length; i++) {
            bw.write(visited[i]+"\n");
        }
        bw.flush();
        bw.close();
    }

    static void dfs(int p, int[] visited) {
        for (int node : list.get(p)) {
            if(visited[node] == 0){
                visited[node] = p;
                dfs(node, visited);
            }
        }
    }

    static void bfs(int root, int[] visited) {
        Queue<Integer> queue = new LinkedList<>();
        queue.add(root);
        while (!queue.isEmpty()) {
            Integer poll = queue.poll();
            for (int node : list.get(poll)) {
                if(visited[node] == 0){
                    visited[node] = poll;
                    queue.add(node);
                }
            }
        }
    }
}

처음에는 '이분 그래프'라는 개념을 처음 접해봐서 이분 그래프 개념 자체를 이해하는 데 애먹었다. 친절한 챗GPT의 도움을 받아 이해가 안되는 걸 물어가며 차근차근 이해했다. 그러고나서 정점을 서로 다른 그룹에 나누는 아이디어까지는 생각보다 쉽게 떠올렸는데 자꾸 시간초과가 났다. boolean으로 된 visited 배열과 각각의 정점을 다른 그룹에 담을 List 배열까지 만들었더니 이게 시간을 많이 잡아먹었었던 것 같다. visited 배열을 boolean이 아니라 int 배열로 선언하고 1 혹은 -1로 그룹을 나누어서 구현했더니 통과했다!

그리고 나는 단순히 1부터 V까지 정점이 있으니까 bfs든 dfs든 1부터 시작해서 순회하면 되겠지~ 생각했는데 중간에 자꾸 틀렸다고 나오더라... 그래서 정점들을 전부 순회하는 for문 안에서 bfs와 dfs를 돌도록 수정했더니 통과할 수 있었다. 중간에 연결되지 않은 정점과 간선도 있다는 것 명심!! 나만 하는 실수인가?

아래는 bfs와 dfs로 구현한 전체 코드이다. 메모리와 시간은 둘 다 비슷하게 소요된다. (dfs가 아주 근소한 차이로 빠르긴하다.)

import java.io.*;
import java.util.*;

// 1707 - 이분 그래프(Bipartite Graph)
public class Main {
    static int[] visited;
    static boolean isBipartite;

    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        StringTokenizer st;
        int K = Integer.parseInt(br.readLine());
        for (int i = 0; i < K; i++) {
            isBipartite = true;
            st = new StringTokenizer(br.readLine());
            int V = Integer.parseInt(st.nextToken()); // 정점
            int E = Integer.parseInt(st.nextToken()); // 간선
            List<List<Integer>> vertexes = new ArrayList<>();
            for (int j = 0; j < V + 1; j++) {
                vertexes.add(new ArrayList<>());
            }
            for (int j = 0; j < E; j++) {
                st = new StringTokenizer(br.readLine());
                int u = Integer.parseInt(st.nextToken());
                int v = Integer.parseInt(st.nextToken());

                vertexes.get(u).add(v);
                vertexes.get(v).add(u);
            }
            bw.write(isBipartiteGraph(vertexes) ? "YES" : "NO");
            bw.write("\n");
        }

        bw.flush();
        bw.close();
    }

    static boolean isBipartiteGraph(List<List<Integer>> graph) {
        visited = new int[graph.size()];
        boolean result = false;
//        return bfs(graph);
  // bfs로 풀 때는 아래를 주석처리하면 된다
        for (int i = 1; i < graph.size(); i++) {
            if (visited[i] == 0) {
                visited[i] = 1;
                result = dfs(graph, i);
                if(!result){
                    return result;
                }
            }
        }

        return result; 
    }


    static boolean bfs(List<List<Integer>> graph) {
        Queue<Integer> queue = new LinkedList<>();

        for (int k = 1; k < graph.size(); k++) {
            if (visited[k] == 0) {
                queue.add(k);
                visited[k] = 1;
            }
            while (!queue.isEmpty()) {
                int vertex = queue.poll();
                int currentGroup = visited[vertex];
                int oppositeGroupIndex = currentGroup * -1;

                for (int i : graph.get(vertex)) {
                    if (visited[i] != 0) {
                        // i들이 같은 그룹에 없어야함! 같은 그룹에 있는지 확인
                        if (visited[i] == currentGroup) {
                            return false;
                        }
                    } else {
                        // 다른 그룹에 넣어야함!
                        visited[i] = oppositeGroupIndex;
                        queue.add(i);
                    }
                }
            }
        }

        return true;
    }


    static boolean dfs(List<List<Integer>> graph, int index) {
        int currentGroup = visited[index];
        int opposite = currentGroup * -1;

        for (int v : graph.get(index)) {
            if (visited[v] != 0) {
                if (visited[v] == currentGroup) {
                    return false;
                }
            } else {
                visited[v] = opposite;
                if (!dfs(graph, v)) {
                    return false;
                }
            }
        }
        return true;
    }
}

원본 백준 문제 페이지는 링크로 남겨두기!

처음에는 단순히 벽을 부술 수 있는 기회를 같이 저장해서 이동한 위치가 0이거나 혹은 1일 때 (벽을 부술)기회가 있으면 큐에 넣고~ 아니면 말고~ 로 풀면 될 거라고 생각했다. 실제로 주어진 테스트 케이스 2개는 통과했고. 하지만 역시 채점 결과는.... 그런데 아무리 생각해도 반례가 생각이 안 나서 구글에 검색해보니 친절하게 반례를 적어놓은 블로그를 발견했다. https://dev-note-97.tistory.com/35

결국 결정적인 아이디어는 벽을 부순 경우와, 벽을 부수지 않은 경우의 거리를 따로 기록해야한다는 것이었다. 그래서 visited 배열을 map 배열에 한 차원을 더해서 벽을 부순 경우와 부수지 않은 경우를 따로 기록하도록 코드를 다시 작성하였다.

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.LinkedList;
import java.util.Queue;
import java.util.StringTokenizer;

// 2206 - 벽 부수고 이동하기
public class Main {
    static int[] dx = {1, -1, 0, 0};
    static int[] dy = {0, 0, 1, -1};
    static BufferedWriter bw;
    static int N;
    static int M;

    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        bw = new BufferedWriter(new OutputStreamWriter(System.out));
        StringTokenizer st = new StringTokenizer(br.readLine());
        N = Integer.parseInt(st.nextToken()); 
        M = Integer.parseInt(st.nextToken()); 
        int[][] map = new int[N][M];
        int[][][] visited = new int[N][M][2];

        for (int i = 0; i < N; i++) {
            String input = br.readLine();
            for (int j = 0; j < M; j++) {
                map[i][j] = input.charAt(j) - '0';
            }
        }
        br.close();
        Queue<int[]> queue = new LinkedList<>();
        queue.offer(new int[]{0, 0, 1}); // 좌표와 벽을 부술 수 있는 개수(기회)
        visited[0][0][1] = 1;
        while (!queue.isEmpty()) {
            int[] poll = queue.poll();
            int x = poll[0];
            int y = poll[1];
            int chance = poll[2];

            for (int i = 0; i < 4; i++) {
                int nx = x + dx[i];
                int ny = y + dy[i];
                if (isValid(nx, ny)) {
                    // 다음 위치가 0이면 현재 chance에 해당하는 visited 배열의 그 위치의 방문 여부를 확인한다.
                    if (map[nx][ny] == 0 && visited[nx][ny][chance] == 0) {

                        // 현재 chance에 해당하는 visited 배열에 거리 + 1을 한 후 큐에 넣는다.
                        queue.offer(new int[] {nx, ny, chance});
                        visited[nx][ny][chance] = visited[x][y][chance] + 1;

                        // 다음 위치가 1이면 chance가 0인 경우(벽을 부순 상태)의 방문 배열을 확인한다.
                    } else if (map[nx][ny] == 1 && visited[nx][ny][0] == 0) {
                        // 기회가 남아 있으면
                        if (chance == 1) {
                            // 다음 위치에 해당하는 벽을 부순 경우의 방문 배열 = 현재 방문 배열의 거리 + 1
                            visited[nx][ny][0] = visited[x][y][1] + 1;
                            queue.offer(new int[] {nx, ny, 0});
                        }
                    }
                }
            }
        }
        // 방문배열 마지막이 둘 다 0이면 도달하지 못한 것이니 -1 => 
        // 둘 중 하나가 0인 경우에는 0이 아닌 것이 최종 거리 => 
        // 둘 다 0이 아닌 경우에는 둘 중 더 작은 것이 최단 거리
        int result = visited[N-1][M-1][0] == 0 && visited[N-1][M-1][1] == 0 ? -1 :
                visited[N-1][M-1][0] == 0 ? visited[N-1][M-1][1] :
                        visited[N-1][M-1][1] == 0 ? visited[N-1][M-1][0] :
                                Math.min(visited[N-1][M-1][0], visited[N-1][M-1][1]);

        bw.write(String.valueOf(result));
        bw.flush();
        bw.close();
    }

    static boolean isValid(int x, int y) {
        return x >= 0 && x < N && y >= 0 && y < M;
    }
}

그런데 코드를 실행해보니 아래와 같은 에러가 발생했다.

에러로그 엔티티의 경우 로그 데이터 객체를 생성자로 받아서 생성되고, 생성된 엔티티가 save되는 시점에 id가 자동으로 생성되도록 되어있었다. 그래서 내가 가짜로(?) 생성한 엔티티에는 id가 아직 존재하지 않고 컨트롤러 테스트의 결과에는 id도 존재해야 하므로 에러가 나는 것은 당연했다.

그런데 문제는... 어떻게 id를 넣어주느냐 하는 것이었다.

로그 데이터 객체는 다른 곳에서도 사용되기 때문에 일관성을 위해 엔티티에서는 로그 데이터를 그대로 받아서 엔티티를 생성하는 방법만을 가져가고 싶었다. 테스트 코드 때문에 억지로 id를 파라미터로 받는 생성자를 만들고 싶지는 않았다. 마찬가지로 builder를 만들거나 setter를 추가하고 싶지도 않았다.

테스트 코드 때문에 구조를 바꾸기는 싫어서 열심히 구글링을 하다가 아래와 같이 ReflectionTestUtils를 사용하는 방법을 찾아냈고 결과적으로 잘 동작했다!

1. JAVA 코드로 Producer 만들기

public class ProducerJAVA {
    // 로그 설정
    private static final Logger log = LoggerFactory.getLogger(ProducerDemoKeys.class.getSimpleName());

    public static void main(String[] args) {
        // create Producer Properties
        Properties properties = new Properties();

// 부트스트랩 서버 설정 
properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092"); // properties.setProperty( "bootstrap.servers", "127.0.0.1:9092"); 로도 설정 가능

// Key serializer 설정  
properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // properties.setProperty("key.serializer", StringSerializer.class.getName()); 로도 설정 가능

// Value serializer 설정        
properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // properties.setProperty("value.serializer", StringSerializer.class.getName()); 로도 설정 가능

        // create the producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        String topic = "demo_java"; // 토픽 설정

        for (int j = 0; j < 3; j++) {
            for (int i = 0; i < 10; i++) {
                String key = "id_" + i; // 키 설정 (생략 가능)
                String value = "hello world " + i; // value 설정

                ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
                producer.send(record,
                        (metadata, exception) -> { // 메시지 전송 callback
                            if(exception == null) {
                                log.info("key: {}", key);
                                log.info("partition: {}", metadata.partition());
                            } else {
                                log.error("Error!! ", exception);
                            }
                        });
            }
        }

        producer.flush(); // 카프카 메세지 전송은 비동기 방식이기 때문에 플러시 안해주면 전송하기 전에 프로그램 끝나버림
        producer.close();
    }
}

2. JAVA 코드로 Consumer 만들기

public class ConsumerJAVA {
    // 로그 설정
    private static final Logger log = LoggerFactory.getLogger(ConsumerDemo.class.getSimpleName());

    public static void main(String[] args) {
        String groupId = "my-java-application"; // 컨슈머 그룹 ID 설정
        String topic = "wikimedia.recentchange.connect"; // 토픽 설정

        // create Consumer Properties
        Properties properties = new Properties();

        // 부트스트랩 서버 설정
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");

        // Key deserializer 설정
        properties.setProperty("key.deserializer", StringDeserializer.class.getName());

        // value deserializer 설정
        properties.setProperty("value.deserializer", StringDeserializer.class.getName());

        // 컨슈머 그룹 ID 설정
        properties.setProperty("group.id", groupId);

        // offset reset 설정
        // latest -> 지금부터 보낸 메세지만 읽겠다 / earliest -> 현재 컨슈머 그룹의 커밋된 오프셋을 찾지 못할 때 가장 처음의 오프셋부터 메시지를 읽겠다
        properties.setProperty("auto.offset.reset", "earliest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(topic)); // 구독할 토픽 설정. 복수 개의 토픽 설정 가능함.

        // 메시지 폴링 루프
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000)); // 최대 1000 밀리초만큼 기다리면서 브로커로부터 메시지를 가져옴
            for (ConsumerRecord<String, String> record : records) {
                log.info("Key: {} , Value: {} ", record.key(), record.value());
                log.info("Partition: {} , Offset: {} ", record.partition(), record.offset());
            }
        }
    }
}

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic third_topic --group 그룹명으로 그룹을 지정하면 된다.

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic --group my-first-group

아래와 같이 kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list 명령어를 통해 컨슈머 그룹 목록을 조회하여 my-first-group 그룹이 생성된 것을 확인할 수 있다.

--describe --group 그룹명 옵션을 이용하면 더 자세한 정보를 확인할 수도 있다.

2. Consumer Group과 Offset

앞선 포스팅에서 컨슈머 콘솔에 --from-beginning 옵션을 주게 되면 해당 토픽의 처음부터 메시지를 읽어온다고 설명한 바 있다. 마찬가지로 컨슈머 그룹을 설정한 후 --from-beginning 옵션을 주게 되면 토픽의 처음부터 메시지를 읽어온다.

이번에는 동일하게 my-first-group 그룹에 속한 두번째 컨슈머 콘솔을 실행하면서 --from-beginning 옵션을 전달해보자.

그러면 두번째 컨슈머는 메시지를 처음부터 읽어오지 못하는 것을 확인할 수 있을 것이다. 그 이유는 카프카는 --from-beginning 옵션이 전달되어도 그룹의 일부로 커밋된 offset이 없는 경우에만 처음부터 읽어오기 때문에 동일한 그룹 이름으로 또 --from-beginning 옵션을 줘도 처음부터 읽어오지 않는 것이다.

offset이란?

파티션 내에서 메시지의 순서를 나타내는 고유한 정수 값으로 offset을 commit한다는 것은 컨슈머가 메시지를 읽고 처리한 후에, 메시지를 마지막으로 어디까지 읽었는지 저장하는 것이다. Kafka는 offset을 각 컨슈머 그룹 별로 관리한다.

참고로 컨슈머 그룹에 커밋된 offset이 존재하는 경우에 처음부터 읽어오지 않는다는 것은 결국 해당 컨슈머 그룹의 offset을 초기화시키면 메시지를 처음부터 읽는다는 것이다. 아래 명령어를 통해 offset을 reset할 수 있다.

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-first-group --reset-offsets --to-earliest --topic first_topic --execute

offset을 초기화시킨 후 다시 해당 그룹에 --from-beginning 옵션을 주면 메시지를 처음부터 읽어오는 것을 확인할 수 있다.

3. Consumer group과 파티션

컨슈머 그룹과 오프셋의 관계를 살펴보았으니 이번에는 컨슈머 그룹과 파티션의 관계를 살펴보자.

컨슈머 그룹과 토픽의 파티션의 관계는 어떻게 될까?

컨슈머 그룹 내의 컨슈머는 해당 토픽 내의 파티션을 나눠가진다. 만약 컨슈머의 수가 파티션의 수보다 작으면 하나의 컨슈머가 1 개 이상의 파티션을 담당하게 되고, 컨슈머의 수와 파티션의 수가 같으면 하나의 컨슈머 당 하나의 파티션을 담당하게 된다. 그리고 만약 컨슈머의 수가 파티션의 수보다 크면 담당하는 파티션 없이 놀고 있는 컨슈머가 존재하게 되는 것이다.

예시를 통해 확인해보자.

1) 파티션이 5개인 'multiple_partiton_topic' 생성

2) 'multiple_partiton_group' 그룹에 속한 하나의 컨슈머 생성

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic multiple_partiton_topic --group multiple_partiton_group -- 컨슈머 생성

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group multiple_partiton_group --describe -- 컨슈머 그룹에 대한 정보 확인

--describe 옵션을 통해 해당 컨슈머 그룹에 대한 정보를 조회한 결과, console-consumer-da1de9ad-ce7d-4fb7-8113-a269fd8d9d11라는 ID를 가진 하나의 컨슈머가 5개의 파티션(0,1,2,3,4)을 담당하고 있는 것을 확인할 수 있다.

3) 'multiple_partiton_group' 그룹에 속한 컨슈머를 추가로 2개 생성(total 3개)

컨슈머를 추가로 2개 더 생성한 후 컨슈머 정보를 조회한 결과이다.

console-consumer-198de97a-792c-4cfc-bd5a-62b664df4ce7가 파티션 0, 1을 담당하고 console-consumer-d112d1d7-3282-4d67-afd7-00cefc51a65d가 파티션 2, 3을 담당하고 console-consumer-da1de9ad-ce7d-4fb7-8113-a269fd8d9d11가 파티션 4를 담당하고 있는 것을 확인할 수 있다.

4) 'multiple_partiton_group' 그룹에 속한 컨슈머를 추가로 2개 생성(total 5개)

multiple_partiton_group 그룹의 컨슈머를 2개 더 생성하여 컨슈머의 수가 토픽의 파티션 수인 5개가 되었다. 이제는 각 컨슈머가 하나의 파티션을 담당하게 되었다.

5) 'multiple_partiton_group' 그룹에 속한 컨슈머를 추가로 2개 생성(total 7개)

multiple_partiton_group 그룹의 컨슈머를 2개 더 생성하여 컨슈머의 수는 파티션의 수보다 많은 7개가 되었다. 참고로 추가로 컨슈머를 더 생성해도 하나의 파티션은 하나의 컨슈머에만 속한다. 하나의 파티션을 여러 개의 컨슈머가 담당하지는 않는다.

아래는 컨슈머를 2개 더 추가한 후에 다시 컨슈머 그룹의 정보를 조회한 결과이다. 여전히 하나의 컨슈머는 하나의 파티션을 담당하고 있는 것을 확인할 수 있다. 하지만 자세히 보면 4번에서 조회한 파티션 - 컨슈머ID 조합과 현재의 조합이 달라진 것을 알 수 있다.

그 이유는 컨슈머 그룹에 컨슈머가 새로 추가되거나 나가면(토픽에 새로운 파티션이 추가될 때도) 리밸런싱이 일어나기 때문이다. 파티션을 컨슈머에게 재할당하는 것이다.(참고로 리밸런싱에도 여러 전략이 존재한다. 여기서 설명하기에는 너무 길어지므로 생략...)

처음에는 '연속된 수의 합 중 가장 큰 값'을 찾는 문제니까 n 개가 연속된 합들을 구할 때 이전까지의 값을 활용(dp...)하고, 그렇게 구한 연속합 중 큰 값을 구하면 되겠다^^! 라고 생각했었다. 그러나 결과는 메모리 초과, 2차 시도는 시간 초과... 아래가 2차 시도 코드이다.

public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        int n = Integer.parseInt(br.readLine());
        StringTokenizer tokenizer = new StringTokenizer(br.readLine());
        br.close();
        int max = -1001;
        int[] values = new int[n + 1];
        int[] sums = new int[n + 1];
        for (int i = 0; i < n; i++) {
            values[i + 1] = Integer.parseInt(tokenizer.nextToken());
            sums[i + 1] = values[i + 1];
            max = Math.max(max, values[i + 1]);
        }

        for (int i = 2; i < n + 1; i++) {
            for (int j = 1; j + i -1 <= n; j++) {
                sums[j] =  sums[j] + values[i+j-1];
                max = Math.max(max, sums[j]);
            }
        }

        bw.write(String.valueOf(max));
        bw.flush();
        bw.close();
    }

막연히 dp는 이전 값들을 배열에 저장해서 사용하는 것! 이라는 것에 꽂혀서 이런 생각을 했던 것 같다. 결국 다른 사람들의 풀이를 보고 아하.. 이전까지의 연속합의 자신의 값을 더한 값과 현재 자신의 값 중 큰 값을 취하는 거구나ㅠㅠ 깨닫고 다시 코드를 짰다. 아래가 그 코드이고 결과는 성공! 휴... dp 어렵다,,,

import java.io.*;
import java.util.StringTokenizer;

// 1912 - 연속합
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        int n = Integer.parseInt(br.readLine());
        StringTokenizer tokenizer = new StringTokenizer(br.readLine());
        br.close();

        int max = Integer.parseInt(tokenizer.nextToken());
        int sum = max;
        for (int i = 1; i < n; i++) {
            int input = Integer.parseInt(tokenizer.nextToken());
            sum = Math.max(input, sum + input);
            max = Math.max(sum, max);
        }

        bw.write(String.valueOf(max));
        bw.flush();
        bw.close();
    }

}

1. Producer & Consumer 콘솔 실행

1) producer와 consumer 실행 명령어를 각각 다른 터미널 창에서 입력한다.

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic -- producer 실행

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic -- consumer

2) producer 콘솔 창에서 메세지를 입력한다.

3) 아래와 같이 consumer 콘솔에 producer가 보낸 메세지가 출력된다.

2. 메시지를 처음부터 Consume하기

위에서 설명한 명령어로 consumer 콘솔을 실행하면 실행한 이후에 전달된 메시지만을 소비한다. 만약에 메세지를 맨 처음부터 읽고 싶다면 --from-beginning 옵션을 사용하면 된다. 아래와 같이 --from-beginning를 추가하여 명령어를 입력하면 컨슈머 콘솔을 실행하기 전에 잘못 보낸(...) 메세지까지 읽을 수 있다.

3. 메시지를 Key와 함께 Produce하기

--property parse.key=true --property key.separator 옵션을 이용해 메세지를 보낼 때 키를 설정하여 전송할 수도 있다. --property key.separator= 다음에 오는 값은 키와 값(메시지)을 구분하는 구분자가 된다.

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic --property parse.key=true --property key.separator=:

예시와 같이 구분자를 :로 설정하고 메시지를 보내보자.

그러면 consumer에서는 값만 전달받고 있는 것을 확인할 수 있다.

** 그렇다면 메시지를 전송할 때 Key는 어떤 역할을 하기에 필요한 것일까? **

1) 파티셔닝 & 논리적 그룹 Kafka는 키를 해시하여 파티션을 결정한다. 이를 통해 같은 키를 가진 메시지는 항상 같은 파티션으로 전송되므로 메시지 키를 사용하면 특정 파티션에 메시지를 보낼 수 있다. 그리고 이를 이용하여 논리적인 그룹으로 메시지를 묶을 수도 있다. 예를 들어서 user의 ID를 키 값으로 사용하면 같은 user의 메시지는 항상 같은 파티션으로 전송되는 것이다.

2) 키값에 따른 순서 보장 동일한 키를 가진 메시지는 동일한 파티션에 저장되기 때문에, 이 파티션에서의 메시지 순서가 보장된다. 키별로 순서를 보장하고자 할 때 유용하다.

4. 메시지를 acks 설정과 함께 Produce하기

메시지를 전달할 때 *--producer-property acks= * 옵션을 줌으로써 acks 모드를 설정할 수 있다. 설정 값으로는 acks=0, acks=1, acks=all(acks=-1)이 있다.

 kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first_topic --producer-property acks=all -- acks 설정

acks란? 메시지를 브로커에 전송할 때의 확인(acknowledgement) 모드

• acks=0: 프로듀서는 브로커로부터 어떤 응답도 기다리지 않는다. 메시지가 성공적으로 전달되었는지 알 수 없기 때문에 데이터 손실 가능성이 있다.

• 속도는 가장 빠르지만 신뢰성이 낮음.

• acks=1: 리더 브로커가 메시지를 처리한 후 Producer에게 응답한다. 어느 정도의 신뢰성은 보장되지만 리더 브로커에 장애가 있을 경우 데이터 손실 가능성이 존재한다.
• acks=all (또는 acks=-1): 리더 브로커 및 리더와 동기화된 replica가 모두 메시지를 처리한 후 프로듀서에게 응답한다. 데이터 손실의 가능성이 거의 없고 가장 신뢰성이 높지만, 속도가 느려질 수 있다.

5. 기타 Producer & Consumer CLI 명령어

1) Producer CLI 명령어

• kafka-console-producer.sh --bootstrap-server localhost:9092 --topic non_existing_topic: 존재하지 않는 토픽에 메시지를 보낼 수 있다. (여기서 non_existing_topic의 자리에는 지금은 존재하지 않지만 produce와 동시에 생성하고자 하는 토픽 이름을 입력) 이렇게 토픽을 생성한 경우에는 해당 토픽의 파티션이 server.properties 파일의 num.partitions 값만큼 생성된다.

2) Consumer CLI 명령어

display key, values and timestamp in consumer

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first_topic --formatter kafka.tools.DefaultMessageFormatter --property print.timestamp=true --property print.key=true --property print.value=true --property print.partition=true --from-beginning : 메시지 format 옵션을 줄 수 있다. 위와 같은 옵션을 주는 경우에는 메시지(value)와 key, partition, timestamp까지 함께 보여준다.

import java.io.*;
import java.util.StringTokenizer;

// 2346 풍선 터뜨리기
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        int n = Integer.parseInt(br.readLine());
        int[] arr = new int[n];
        StringTokenizer tokenizer = new StringTokenizer(br.readLine());
        for (int i = 0; i < n; i++) {
            arr[i] = Integer.parseInt(tokenizer.nextToken());
        }
        int popNext = 0;
        int popCount = 1;
        bw.write((popNext + 1) + " ");

        while (popCount < arr.length) {
            int popOrder = arr[popNext];
            arr[popNext] = 0;

            if (popOrder > 0) {
                for (int i = 0; i < popOrder; i++) {
                    int index = (i + 1 + popNext) % arr.length;

                    if (arr[index] == 0) {
                        popOrder++;
                    } else if (i == popOrder - 1) {
                        popNext = index;
                        popCount++;
                        bw.write((popNext + 1) + " ");
                    }
                }
            } else { // If the popOrder is negative
                for (int i = 0; i > popOrder; i--) {
                    int index = popNext + (i - 1) % arr.length < 0 ? arr.length + popNext + (i - 1) % arr.length : popNext + (i - 1) % arr.length;
                    if (arr[index] == 0) {
                        popOrder--;
                    } else if (i == popOrder + 1) {
                        popNext = index;
                        popCount++;
                        bw.write((popNext + 1) + " ");
                    }
                }
            }
        }

        br.close();

        bw.flush();
        bw.close();
    }
}

그런데 스택, 큐, 덱 단계에서 메모리 제한이나 시간 제한을 둔 문제들이 꽤 나왔고 24511 queuestack에서 시간 초과로 많은 빠꾸... 아니 고배를 마셨기에 복기해보려고 한다.

처음에는 그냥 단순히 배열로 풀어보려다 -> 시간초과 배열을 2개 써서? 2차원 배열로 간다 -> 시간초과 배열을 0(queue)만 저장 -> 시간초과 0(queue)인 값의 갯수까지 저장해서 그만큼만 for문 -> 시간초과 ......

결국 구글링해서 Deque를 쓰면 된다는 걸 깨달았다. 사실 나는 스택,큐,덱에서 덱이 뭔지도 몰랐던 사람인지라 Deque가 한 번에 생각나지 않았던 것 같다. 막상 깨닫고 나니 어렵지 않은 문제였다! Deque를 잘 기억해두자

import java.io.*;
import java.util.*;

// 24511 queuestack
public class Main {
    public static void main(String[] args) throws IOException {
        BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(System.out));
        int n = Integer.parseInt(br.readLine());
        Deque<Integer> deque = new ArrayDeque<>();
        StringTokenizer tokenizer = new StringTokenizer(br.readLine());
        StringTokenizer tokenizer2 = new StringTokenizer(br.readLine());

        for (int i = 0; i < n; i++) {
            String value = tokenizer2.nextToken();
            if (tokenizer.nextToken().equals("0")) {
                deque.add(Integer.parseInt(value));
            }
        }

        n = Integer.parseInt(br.readLine());
        tokenizer = new StringTokenizer(br.readLine());
        for (int i = 0; i < n; i++) {
            int input = Integer.parseInt(tokenizer.nextToken());
            deque.offerFirst(input);
            bw.write(deque.pollLast() + " ");
        }

        br.close();

        bw.flush();
        bw.close();
    }
}

1. kafka 설치

brew를 사용하는 경우 아래 명령어로 kafka를 설치한다.

brew install kafka

설치를 끝내고 kafka가 설치된 디렉토리의 bin 폴더로 가보면 아래와 같이 여러 개의 sh 파일들이 설치되어 있는 것을 확인할 수 있다.

2. kafka 실행

kafka를 실행하는 방법에는 zookeeper와 함께 실행하는 방법과 zookeeper 없이 실행하는 방법이 존재한다.

2-1. zookeeper와 함께 kafka 실행

zookeeper-server-start.sh (본인 pc의 카프카 config 경로)/zookeeper.properties 명령어로 zookeeper를 실행한다.

zookeeper-server-start.sh (본인 pc의 카프카 config 경로)/zookeeper.properties

-- 명령어 예시
zookeeper-server-start.sh /opt/homebrew/Cellar/kafka/3.7.0/libexec/config/zookeeper.properties

만약에 zookeeper-server-start.sh 로 실행이 되지 않는다면 환경변수 설정을 하거나 카프카 bin 폴더 안에서 ./zookeeper-server-start.sh 로 실행하면 작동할 것이다.

kafka-server-start.sh (본인 pc의 카프카 config 경로)/server.properties 명령어로 kafka server를 실행한다.

kafka-server-start.sh (본인 pc의 카프카 config 경로)/server.properties

-- 명령어 예시
kafka-server-start.sh /opt/homebrew/Cellar/kafka/3.7.0/libexec/config/server.properties

2-2. zookeeper 없이 kafka 실행(KRaft 모드)

1) 아래 명령어로 uuid 생성

kafka-storage.sh random-uuid -- 생성된 uuid가 출력된다. ex) 8OrZqVa1Sp2G_OAujOnKAA

2) 카프카가 데이터를 저장하는 경로를 format 아래 명령어로 kraft-combined-logs(카프카가 로그 데이터를 저장하는 경로)를 format하여 초기화한다. 참고로 지정된 UUID는 클러스터 ID로 설정된다.

kafka-storage.sh format -t <위에서 생성한 uuid> -c  /opt/homebrew/Cellar/kafka/3.7.0/libexec/config/Kraft/server.properties
-- kafka-storage.sh format -t 8OrZqVa1Sp2G_OAujOnKAA -c  /opt/homebrew/Cellar/kafka/3.7.0/libexec/config/Kraft/server.properties

3) 카프카 서버 실행

kafka-server-start.sh (본인 pc의 카프카 config 경로)/server.properties 명령어로 kafka server를 실행한다.

kafka-server-start.sh (본인 pc의 카프카 config 경로)/server.properties 

-- 데몬으로 실행할 경우
kafka-server-start.sh -daemon (본인 pc의 카프카 config 경로)/server.properties 

참고로 앞으로 카프카는 주키퍼를 제거하는 방향으로 나아갈 것이므로 주키퍼 없이 사용하는 것을 권장한다고 한다.

• 혹시나하고 윈도우에서 실행해봤는데 KRaft 모드로 문제 없이 실행된다!

3. Topic 생성 및 조회하기

1) 토픽 생성

kafka-topics.sh --bootstrap-server localhost:9092 --topic 생성할 토픽명 --create 명령어로 토픽을 생성한다. 아래는 first_topic이라는 이름의 토픽을 생성한 결과이다. localhost:9092 부분은 부트스트랩 서버의 port를 변경했거나 다른 브로커를 부트스트랩 서버로 설정하고 싶으면 그에 맞게 변경해주면 된다.

2) 토픽 조회

kafka-topics.sh --bootstrap-server localhost:9092 --list 명령어로 토픽을 조회할 수 있다.

• 부트스트랩 서버 (Bootstrap Servers)란? 부트스트랩 서버는 Kafka 클라이언트(프로듀서 또는 컨슈머)가 초기 연결을 설정하는 데 사용되는 브로커의 목록으로 부트스트랩 서버의 주요 역할은 클라이언트가 클러스터의 메타데이터(ex. 브로커 목록, 토픽 정보 등)를 가져올 수 있도록 하는 것이라고 한다. 예를 들어 내 pc에 localhost:9092, localhost:9093 localhost:... 와 같이 여러 개의 포트에 해당하는 여러 개의 브로커가 존재할 때, bootstrap-server를 통해 모든 브로커의 정보를 받아올 수 있는 것이다.

3) 기타 토픽 CLI 명령어 옵션

• kafka-topics.sh --bootstrap-server localhost:9092 --topic first_topic --describe : 지정한 토픽에 대한 세부 정보(파티션, 복제본, ISR) 등의 정보 표시

• kafka-topics.sh --bootstrap-server localhost:9092 --topic first_topic --delete : 토픽 삭제

• kafka-topics.sh --bootstrap-server localhost:9092 --topic first_topic --create --partitions 3 : 생성할 토픽에 3개의 파티션을 생성

• kafka-topics.sh --bootstrap-server localhost:9092 --topic first_topic --create --partitions 3 --replication-factor 2 : 각 파티션에 2개의 레플리카 생성