Data Stream 예제(Subscribe)

Spring Cloud Stream 기반으로 처리되는 구독(Subscribe) 예제 5종을 제공하며 메시지 브로커의 메시지를 전달 받아 다양한 방법으로 처리한다.

메시지 브로커에서 온도 및 습도 정보를 구독하여 관련 데이타를 가져온 다음 Websocket을 이용하여 화면에 구글차트로 결과를 보여준다.

spring:
  cloud:
    stream:
      bindings:
        sensorThermo-in-0:
          destination: sensor
          binder: rabbit
      function:
        definition: sensorThermo

수집된 온습도 데이터에 대해 실시간으로 차트에 결과를 보여준다.

메시지 브로커에서 라인기반 파일 정보를 구독하여 관련 데이타를 가져온 Websocket을 이용하여 화면에 결과를 보여준다.

spring:
  cloud:
    stream:
      bindings:
        fileLine-in-0:
          destination: line-topic
          binder: kafka
      function:
        definition: fileLine

전달된 텍스트 데이타에 대해 실시간으로 화면에 결과를 보여준다.

Elastic Search가 상용화 되면서 해당 OSS(Open Source Software)에서 분기한 OpenSearch를 이용하면 편리하게 데이터를 수신하여 시각화 할 수 있다. 이때 메시지 서버와 연계는 Logstash를 이용하여 처리한다.

Logstash는 로그 형태의 데이타를 OpenSearch로 쉽게 전달할 수 있다.

* input > group_id : consumer 그룹 지정
* input > topics : egov-logs (Kafka 토픽 명)
* input > decorate_events ⇒ 이걸 설정해야 @metadata 같이 옵션 사용가능
* filter > grok 플러그인 : 메시지 패턴 적용
* output > index : YYYY.MM.dd로 일별 구분

input {
  kafka {
    bootstrap_servers => "192.168.100.50:9092"
    group_id => "logstash"
    topics => ["egov-logs"]
    consumer_threads => 1
    decorate_events => true
  }
 
}
 
filter {
  grok {
    match => {
	  "message" => "\[%{TIMESTAMP_ISO8601:timestamp}] \[%{DATA:thread}] %{LOGLEVEL:logLevel} %{JAVACLASS:packageClass} - %{GREEDYDATA:msg}"
    }
  }
}
 
output {
 
  stdout {}
 
  opensearch {
    hosts => ["https://192.168.100.50:9200"]
    index => "egov-logs-%{+YYYY.MM.dd}"
    user => "admin"
    password => "admin"
    ssl => true
    ssl_certificate_verification => false
  }
 
}

메시지 브로커에서 DB CRUD 정보를 구독하여 Websocket을 이용하여 화면에 결과를 보여준다.

spring:
  cloud:
    stream:
      bindings:
        historyDb-in-0:
          destination: db-topic
          binder: kafka
      function:
        definition: historyDb

CRUD에 의해 조회된 이력정보를 구독하여 실시간으로 차트에 출력한다.

메시지 브로커에서 Open API 정보를 구독하여 MongoDB에 저장한다.

spring:
  cloud:
    stream:
      bindings:
        trainApi-in-0:
          destination: api-topic
          binder: kafka
      function:
        definition: trainApi

구독된 데이타는 즉시 MongoDB의 Collection에 저장된다.
MongoDB에서 제공하는 Compass client를 사용하면 손쉽게 Collection 확인이 가능하다.

• Cloud Data Stream 환경설정
• grok 문서
• grok 패턴 정의
• grok 패턴 테스트 URL
• MongoDB Compass