TableView 사용하기

TableView 사용하기 (TableView)

TableView는 토픽의 최신 상태를 키-값 맵처럼 바라볼 수 있게 해주는 기능이에요. 이 문서는 TableView를 구성하는 방법과 리스너를 등록하는 방법을 정리했어요. 트ombstone(null 페이로드) 처리와 forEach/forEachAndListen의 차이도 함께 살펴볼게요.

출처: 문서

본문

클라이언트를 설정한 뒤에는 더 살펴보며 TableView로 작업을 시작할 수 있어요.

TableView 구성 (Configure TableView)

  • Java C++

다음은 TableView를 구성하는 예시예요.

TableView<String> tv = client.newTableViewBuilder(Schema.STRING)
    .topic("my-tableview")
    .create()

loadConf 구성이나 TableViewBuilder API의 사용 가능한 파라미터로 TableView를 커스터마이즈할 수 있어요.

이름 타입 필수? 설명 기본값
topic string 예 TableView의 토픽 이름. N/A
autoUpdatePartitionInterval int 아니요 새로 추가된 파티션을 확인하는 간격. 60 (초)
subscriptionName string 아니요 TableView의 서브스크립션 이름. null

이 기능은 C++ 클라이언트 3.2.0 이상 버전에서 지원돼요.

ClientConfiguration clientConfiguration;
clientConfiguration.setPartititionsUpdateInterval(100);
Client client("pulsar://localhost:6650", clientConfiguration);

TableViewConfiguration tableViewConfiguration{schemaInfo, "test-subscription-name"};
TableView tableView;
client.createTableView("my-tableview", tableViewConfiguration, tableView)

다음 파라미터로 TableView를 커스터마이즈할 수 있어요.

이름 타입 필수? 설명 기본값
topic string 예 TableView의 토픽 이름. N/A
schemaInfo struct 아니요 이 TableView가 받아들일 수 있는 데이터의 스키마를 선언해요. 스키마는 토픽의 스키마와 대조되며, 호환되지 않으면 TableView 생성이 실패해요. N/A
subscriptionName string 아니요 TableView의 서브스크립션 이름. reader-{random string}
partititionsUpdateInterval int 아니요 토픽 파티션 업데이트 간격(초). C++ 클라이언트에서 partititionsUpdateInterval은 같은 클라이언트 안에서 전역이에요. 60

TableView는 null 페이로드를 가진 메시지를 tombstone으로 취급해요 — 그 키는 맵에서 제거돼요. forEach(action)은 현재 맵 스냅샷만 반복하므로, tombstone 처리된 키는 표면에 드러나지 않아요. forEachAndListen(action)은 먼저 현재 비-tombstone 항목에 대해 forEach를 실행한 다음, 액션을 라이브 리스너로 등록해요. 그 후의 모든 업데이트 — tombstone 포함 — 는 action.accept(key, null)으로 리스너에 전달돼요. 삭제에 반응해야 한다면(예: 다운스트림 상태를 정리하기 위해) 리스너에서 value == null을 확인하세요.

리스너 등록 (Register listeners)

forEachAndListen를 사용해 토픽의 기존 메시지와 토픽으로 들어오는 새 메시지 모두에 대한 리스너를 등록하고, forEach를 사용해 모든 기존 메시지에 대해 연산을 수행하도록 지정할 수 있어요.

다음은 TableView로 리스너를 등록하는 예시예요.

  • Java C++
// Register listeners for all existing and incoming messages
tv.forEachAndListen((key, value) ->
    /*operations on all existing and incoming messages*/)

// Register actions for all existing messages
tv.forEach((key, value) ->
    /*operations on all existing messages*/)

// Register listeners for all existing and incoming messages
tableView.forEach([](const std::string& key, const std::string& value) {});

// Register actions for all existing messages
tableView.forEachAndListen([](const std::string& key, const std::string& value) {});

더 알아보기 (Learn more)