kafka 를 설치 하고 외부 네트워크 ip 를 통 해 접근 하도록 설정 합 니 다.

1.kafka 설치
JDK 설치

tar xvf jdk1.8.0_231.tar.gz  -C /usr/local && cd   /usr/local
ln -sv jdk1.8.0_231 jdk 

vim /etc/profile.d/java.sh
    JAVA_HOME=/usr/local/jdk
    PATH=$JAVA_HOME/bin:$PATH

zookeeper 설치(또는 kafka 자체 테이프 사용)

vim /usr/local/kafka/zookeeper/conf/zoo.cfg   
   tickTime=2000
   initLimit=10
   syncLimit=5
   dataDir=/data/zookeeper
   clientPort=2181
   maxClientCnxns=0

   #     zookeeper      
   # server.1=ip1:2888:3888
   # server.2=ip2:2888:3888
   # server.3=ip3:28888:3888

kafka 다운로드 및 kakfa 설치

wget https://archive.apache.org/dist/kafka/0.10.2.1/kafka_2.11-0.10.2.1.tgz

tar xvf kafka_2.11-0.10.2.1.tgz -C /usr/local && cd   /usr/local
ln -sv kafka_2.11-0.10.2.1.tgz kafka

kafka 시작 메모리 수정

vim /usr/local/kafka/bin/kafka-server-start.sh
  export KAFKA_HEAP_OPTS="-Xmx2G -Xms2G"

kafka 시작 및 정지

/usr/local/kafka/bin/zookeeper-server-start.sh -deamon /usr/local/kafka/conf/zookeeper.properties

/usr/local/kafka/bin/kafka-server-start.sh -deamon /usr/local/kafka/conf/server.properties

/usr/local/kafka/bin/kafka-server-stop.sh /usr/local/kafka/conf/server.properties
/usr/local/kafka/bin/zookeeper-server-stop.sh /usr/local/kafka/conf/zookeeper.properties

kafka 를 단독으로 설치 하 는 정지 방식

/usr/local/zookeeper/bin/zkServer.sh stop|stop

2.kafka 는 외부 네트워크 방문 을 설정 합 니 다(외부 네트워크 방문 이 필요 하 다 면)
전제 조건
내부 네트워크 ip 주 소 를 분석 하 는 도 메 인 이 필요 합 니 다.내부 네트워크 환경 도/etc/hosts 를 설정 할 수 있 습 니 다.
매개 변수 설정
host.name=kafka.test.com(대응 하 는 도 메 인 이름 해석 은 네트워크 ip 로 풀 어야 합 니 다)고 버 전이 폐기 되 었 습 니 다.낮은 버 전 0.10.2.1 은 listeners 속성 이 설정 되 지 않 았 을 때 만 사용 할 수 있 으 며,listeners 속성 으로 대체 되 었 습 니 다.broker 를 나타 내 는 hostname
advertised.listeners=PLAINTEXT:/kafka.test.com:9092(높 은 버 전 으로 host.name 대신 advertised.listeners 를 설정 하여 host.name 을 설정 하지 않 고)zookeeper 에 등록 하여 클 라 이언 트 에 제공 하 는 모니터 를 설정 하고 설정 하지 않 으 면 listeners 를 사용 합 니 다.
advertised.host.name(설정 필요 없 이 참고 만)이 버 려 졌 습 니 다.advertised.listeners 나 listeners 속성 이 설정 되 지 않 았 을 때 만 사 용 됩 니 다.홈 페이지 advertised.listeners 사용 권장
listeners(설정 필요 없 이 참고 만)감청 할 URL 과 프로 토 콜,예 를 들 어 PLAINTEXT://my host:9092,SSL:/:9091 CLIENT:/0.0.0.0.0:9092,REPLICATION:/localhost:9093.이 설정 이 지정 되 지 않 으 면 자바.net.InetAddress.getCanonicalHostName()함수 의 반환 값 을 사용 합 니 다.
브로커 의/etc/hosts 파일 수정
[내부 ip]kafka.test.com
외부 네트워크 액세스 서버 의/etc/hosts 파일 수정
[외부 네트워크 ip]kafka.test.com
3.kafka 소비 테스트
생산자

/usr/local/kafka/bin/kafka-console-producer.sh --broker-list IP:9092 --topic TOPIC

소비자

/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server IP:9092 --topic TOPIC--from-beginning --max-messages 1
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server   IP:9092 --topic TOPIC --from-beginning --max-messages 1

4.logstash 디 버 깅

output {
  stdout { codec => rubydebug { metadata => true }  }
}

5.logstash 에서 kakfa 로 그 를 소비 할 수 없 는 문제 조사
a、topics_pattern 통 배 문제

      topics_pattern=>"prefix-.*" 

b.filter 에서 일치 하 는 규칙 입 니 다.kafka 에서 topic 와 일치 해 야 합 니 다.서로 다른 filebeat 는 서로 다른 logstash 버 전에 대응 하 는 topic 메타 데이터 와 다 를 수 있 습 니 다.이 점 은 주의해 야 합 니 다.

      if [type] =~ "prefix-*" {
          grok { match =>["[type]","^prefix-(?)"] }
       }

      if [kafka][topic] =~ "prefix-*" {
        grok { match => [ "[kafka][topic]", "^prefix-(?.*$)" ]}
      }

       if [@metadata][topic] =~ "prefix-*" {
          grok { match =>["[@metadata][topic]","^prefix-(?)"] }
       }

      if [@metadata][kafka][topic] =~ "prefix-*" {
        grok { match => [ "[@metadata][kafka][topic]", "^prefix-(?.*$)" ]}
      }

외부 네트워크 kakfa 소비 참고:https://www.maiyewang.com/archives/17993