Cài đặt Apache Kafka trên Centos 7

Kafka đã có phiên bản 2.0: https://kafka.apache.org/downloads

1. Về Kafka

Về cơ bản kafka là hệ thống message pub/sub phân tán mà có khả năng scale rất tốt.
Message của kafka được lưu trên đĩa cứng, đồng thời được replicate trong cluster giúp phòng tránh mất dữ liệu.
Kafka có thể hiểu là một hệ thống logging, nhằm lưu lại các trạng thái của hệ thống, nhằm phòng tránh mất thông tin.

Thông tin cụ thể hơn các bạn xem tại đây

2. Cài đặt

Kafka mới được Apache tung ra phiên bản 1.0, nghĩa là phiên bản chính thức, không còn là các phiên bản 0.x nữa. Điều đó có nghĩa Kafka đã trở thành phần mềm khá hoàn hảo rồi.

Trước tiên cần cài đặt Java. Có nhiều cách như tải trên trang chủ (bản mới nhất), hoặc cài từ yum

sudo yum install java-1.8.0-openjdk.x86_64

https://kafka.apache.org/downloads

https://kafka.apache.org/downloads

https://kafka.apache.org/downloads

Tải phiên bản kafka_2.11-1.0.0.tgz ở đây có nghĩa là phiên bản kafka 1.0 chạy trên nền scala 2.11

Giải nén và vứt vào thư mục /opt/kafka

sudo tar -xvf kafka_2.11-1.0.0.tgz

sudo mv kafka_2.11-1.0.0 /opt/kafka

cd /opt/kafka/

Start Zookeeper:

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

Start Kafka:

bin/kafka-server-start.sh config/server.properties

Zookeeper lắng nghe ở port 2181 còn Kafka là port 9092

Có thể tạo service trong init.d để quản lý dễ dàng hơn:

sudo nano /etc/init.d/kafka

#! /bin/bash

### BEGIN INIT INFO
# Provides:  kafka
# Required-Start:
# Required-Stop:
# Default-Start: 2 3 4 5
# Default-Stop: 0 1 6
# Short-Description: kafka service
### END INIT INFO

ZNAME="zookeeper"
KNAME="kafka"

ZCMD="/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties"
KCMD="/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties"
ZPIDFILE="/var/run/$ZNAME.pid"
KPIDFILE="/var/run/$KNAME.pid"
ZLOGFILE="/var/log/$ZNAME.log"
KLOGFILE="/var/log/$KNAME.log"

recursiveKill() { # Recursively kill a process and all subprocesses
  CPIDS=$(pgrep -P $1);
  for PID in $CPIDS
  do
    recursiveKill $PID
  done
  sleep 3 && kill -9 $1 2>/dev/null & # hard kill after 3 seconds
  kill $1 2>/dev/null # try soft kill first
}

function zstart {
  echo "Starting $ZNAME ..."
  if [ -f "$ZPIDFILE" ]; then
    echo "Already running according to $ZPIDFILE"
  else
    /bin/su -m -c "$ZCMD" > $ZLOGFILE 2>&1 &
    PID=$!
    echo $PID > $ZPIDFILE
    echo "Started $ZNAME with pid $PID - Logging to $ZLOGFILE"
  fi
}

function kstart {
  echo "Starting $KNAME ..."
  if [ -f "$KPIDFILE" ]; then
    echo "Already running according to $KPIDFILE"
    exit 1
  else
    /bin/su -m -c "$KCMD" > $KLOGFILE 2>&1 &
    PID=$!
    echo $PID > $KPIDFILE
    echo "Started $KNAME with pid $PID - Logging to $KLOGFILE"
  fi
}

function zstop {
  echo "Stopping $ZNAME ..."
  if [ ! -f $ZPIDFILE ]; then
    echo "Already stopped!"
  else
    PID=`cat $ZPIDFILE`
    recursiveKill $PID
    rm -f $ZPIDFILE
    echo "Stopped $ZNAME"
  fi
}

function kstop {
  echo "Stopping $KNAME ..."
  if [ ! -f $KPIDFILE ]; then
    echo "Already stopped!"
  else
    PID=`cat $KPIDFILE`
    recursiveKill $PID
    rm -f $KPIDFILE
    echo "Stopped $KNAME"
  fi
}

function zstatus {
  if [ -f "$ZPIDFILE" ]; then
    PID=`cat $ZPIDFILE`
    if [ "$(/bin/ps --no-headers -p $PID)" ]; then
      echo "$ZNAME is running (pid : $PID)"
    else
      echo "Pid $PID found in $ZPIDFILE, but not running."
    fi
  else
    echo "$ZNAME is NOT running"
  fi
}

function kstatus {
  if [ -f "$KPIDFILE" ]; then
    PID=`cat $KPIDFILE`
    if [ "$(/bin/ps --no-headers -p $PID)" ]; then
      echo "$KNAME is running (pid : $PID)"
    else
      echo "Pid $PID found in $KPIDFILE, but not running."
    fi
  else
    echo "$KNAME is NOT running"
  fi
}

case "$1" in
  start)
    zstart
    kstart
    ;;
  stop)
    kstop
    zstop
    ;;
  restart)
    $0 stop
    sleep 3
    $0 start
    ;;
  status)
    zstatus
    kstatus
    ;;
  *)
    echo "Usage: /etc/init.d/kafka {start|stop|restart|status}" && exit 1
    ;;
esac

Nhớ chmod +x /etc/init.d/kafka để file có thể thực thi.

Bây giờ có thể sử dụng lệnh service kafka start/stop/restart để quản lý.

Trường hợp start mà kafka bị stop (chỉ lên mỗi zookeeper) thì chỉ cần stop và start lại 1 lần nữa là được.