Chào các bạn,

Trong bài viết này mình sẽ hướng dẫn mọi người cài đặt Apache Kafka trên CentOS 7.

Giới thiệu về Kafka mọi người có thể xem ở đây.

Hướng dẫn cài đặt

Yêu cầu

Vì kafka được phát triển bằng java nên khi chạy cần có JDK.
Mà để chạy được JDK thì cấu hình tối thiểu của VPS/Server phải được 4GB RAM.

Cài đặt JDK

Trên CentOS 64bit, thực hiện các lệnh sau để tải JDK về và giải nén:

# cd /opt/
# wget --no-cookies --no-check-certificate --header "Cookie:
gpw_e24=http%3A%2F%2Fwww.oracle.com%2F;
oraclelicense=accept-securebackup-cookie"
"http://download.oracle.com/otn-pub/java/jdk/8u121-b13/e9e7ea248e2c4826b92b3f075a80e441/jdk-8u121-linux-x64.tar.gz"
# tar xzf jdk-8u121-linux-x64.tar.gz

Đường link: http://download.oracle.com/otn-pub/java/jdk/8u121-b13/e9e7ea248e2c4826b92b3f075a80e441/jdk-8u121-linux-x64.tar.gz trong câu lệnh trên, mỗi lần tải bạn cần phải đăng nhập vào link này: https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html chọn Accept License Agreement để đồng ý với các chính sách của Oracle, sau đó chọn đúng phiên bản JDK cần tải về để lấy link tải.

Trên CentOS 32bit, thực hiện các lệnh sau để tải JDK về và giải nén:

# cd /opt/
# wget --no-cookies --no-check-certificate --header "Cookie: gpw_e24=http%3A%2F%2Fwww.oracle.com%2F; oraclelicense=accept-securebackup-cookie" "http://download.oracle.com/otn-pub/java/jdk/8u121-b13/e9e7ea248e2c4826b92b3f075a80e441/jdk-8u121-linux-i586.tar.gz"
# tar xzf jdk-8u121-linux-i586.tar.gz

Cài đặt JDK:

# cd /opt/jdk1.8.0_121/
# alternatives --install /usr/bin/java java /opt/jdk1.8.0_121/bin/java 2
# alternatives --config java
There is 1 program that provides 'java'. Selection    Command ----------------------------------------------- *+ 1           /opt/jdk1.8.0_121/bin/java Enter to keep the current selection[+], or type selection number: 1

Thiết lập đường dẫn javac và jar.

# alternatives --install /usr/bin/jar jar /opt/jdk1.8.0_121/bin/jar 2
# alternatives --install /usr/bin/javac javac /opt/jdk1.8.0_121/bin/javac 2
# alternatives --set jar /opt/jdk1.8.0_121/bin/jar
# alternatives --set javac /opt/jdk1.8.0_121/bin/javac

Kiểm tra version java đã được cài đặt:

[root@lantqxx jdk1.8.0_121]# java -version
java version "1.8.0_121"
Java(TM) SE Runtime Environment (build 1.8.0_121-b13)
Java HotSpot(TM) 64-Bit Server VM (build 25.121-b13, mixed mode)

Ở chỗ này, nếu bạn gõ lệnh: java -version mà báo lỗi java command not found thì bạn cần phải link file binary java trong thư mục cài đặt ra /usr/bin/.
Gõ lệnh: ln -fs /opt/jdk1.8.0_181/bin/java /usr/bin/java
Thiết lập biến môi trường JAVA_HOME:

# export JAVA_HOME=/opt/jdk1.8.0_121 

Thiết lập biến môi trường JRE_HOME:

# export JRE_HOME=/opt/jdk1.8.0_121/jre

Thiết lập biến môi trường PATH:

# export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:$JAVA_HOME/bin:$JRE_HOME/bin 

Thêm 3 dòng vừa gõ vào file /etc/environment để tự động chạy khi khởi động hệ thống.

Cài đặt Apache Kafka

Bước 1: Tạo user kafka

Để đảm bảo an toàn cho hệ thống thì trong quá trình chạy kafka, ta nên cho ứng dụng này chạy dưới một user riêng, ở đây ta sẽ tạo user kafka.

useradd kafka -m

Tham số -m là để đảm bảo tạo thư mục home cho user này /home/kafka. Chúng ta sẽ cho ứng dụng kafka chạy ở đây.

Đặt mật khẩu cho user vừa tạo:

passwd kafka

Thêm user kafka vừa tạo vào group wheel để có quyền su.

usermod -aG wheel kafka

Giờ ta có thể đăng nhập vào user kafka:

su -l kafka

Bước 2: Tải Kafka

wget www-eu.apache.org/dist/kafka/2.0.0/kafka_2.12-2.0.0.tgz
tar xzf kafka_2.12-2.0.0.tgz --strip 1

Tham số –strip 1 là để giải nén thẳng các file bên trong file nén ra thư mục hiện tại luôn, không phải giải nén ra thành thư mục kafka_2.12-2.0.0.

Gõ lệnh ls -l để xem các file được giải nén ra thành công:

Bước 3: Cấu hình Kafka Server

Mặc định Kafka không cho phép xóa các topic, để có thể xóa được topic đã tạo, ta cần chỉnh sửa file cấu hình Server Kafka.

echo "delete.topic.enable = true" >> config/server.properties

Bước 4: Tạo Systemd Unit Files và khởi động Kafka Server

Kafka sử dụng Zookeeper để quản lý Cluster cho nên ta cũng cần tạo Systemd Unit File cho Zookeeper.

sudo vi /etc/systemd/system/zookeeper.service

Nhập vào các dòng bên dưới và lưu lại:

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=kafka
ExecStart=/home/kafka/bin/zookeeper-server-start.sh /home/kafka/config/zookeeper.properties
ExecStop=/home/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

Tạo Systemd Unit File cho Kafka:

sudo vi /etc/systemd/system/kafka.service

Nhập các dòng bên dưới vào và lưu lại:

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=kafka
ExecStart=/bin/sh -c '/home/kafka/bin/kafka-server-start.sh /home/kafka/config/server.properties > /home/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

Khởi động Zookeeper và Kafka:

sudo systemctl start zookeeper
sudo systemctl start kafka

Kafka mặc định listen port 9092. Còn Zookeeper là 2181.

Bước 5: Test thử

Chúng ta sẽ tạo một Topic có tên là “Test” để nhận messages từ Procuder và gửi messages ra cho Consumer.

Tạo Topic:

~/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic Test

Kafka sẽ nhận input từ cmd mình nhập vào hoặc là từ file. Mặc định thì mỗi dòng sẽ là 1 message.

Procuder gửi message:

~/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Test
This is a message test 1.
This is a message test 2.

Nhấn Ctrl+C để thoát.

Consumer nhận message:

~/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic Test --from-beginning
This is a message test 1.
This is a message test 2.

Nhấn Ctrl+C để thoát.

Bạn có thể kiểm tra bằng cách mở 2 terminal để chạy Procuder gửi message, mở 1 terminal để chạy Consumer nhận message để test.

Bước 6: Cài đặt Multi Broker Cluster

Ở Bước 4 tạo Systemd Unit File và khởi động Kafka Server thì ta có gọi file server.properties, để tạo multi broker cluster thì ta sẽ tạo nhiều file server.properties này.

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

Sửa một số thông tin trong 2 file này:

config/server-1.properties:
broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/tmp/kafka-logs-1

config/server-2.properties:
broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/tmp/kafka-logs-2

Khởi động 2 Server này lên:

bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &

Tạo một topic mới replication factor là 3:

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic Test-cluster

Xem mô tả của topic vừa tạo:

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic Test-cluster

Dòng đầu tiên là hiển thị tổng số Partition, mỗi dòng tiếp theo sẽ là thông tin của 1 partition. Ở đây vì chúng ta khởi tạo topic với 1 partition cho nên chỉ có 1 dòng đi kèm.

• Leader: là broker (node) chịu trách nhiệm đọc và ghi cho partition đã chỉ định. Ở đây ta thấy broker 0 – chính là con Server kafka được ta khởi động lên bằng lệnh ở bước 4 làm leader cho partition 0.
• Replicas: danh sách các broker (node) có sao chép lưu log cho partition này, bất kể broker đó là leader hay là follower. Ở đây ta thấy rằng cả 3 broker (node) đều có lưu thông tin của partition 0. Với broker 0 là leader, broker 1 và 2 là follower.
• Isr: Cho biết thông tin các broker (node) replicas nào đang ở chế độ “in-sync“. Chỉ hiển thị các replicas đang còn sống và kết nối được với leader. Ở đây ta thấy là cả 3 broker (node) đều đang online và các follower đều đang kết nối đến leader.

Như vậy thì ta sẽ thấy rằng, nếu có 1 message gửi đến topic Test-cluster thì nó sẽ được ghi xuống partition 0, và broker 0 sẽ nhận nhiệm vụ đọc và ghi. Nếu mà broker 0 die thì Zookeeper sẽ chọn 1 trong 2 broker còn lại làm leader thay thế và thông báo cho Procuder, Consumer biết để gửi nhận message cho chuẩn.

Bạn có thể thử test gửi nhận message như ở Bước 5, sau đó kill pid của 1 trong 3 broker, rồi chạy lại lệnh xem mô tả của Topic ở trên. Rồi lại chạy Consumer đọc message output xem có bình thường hay không.

Bạn cũng có thể thử tạo topic mới với số partition lớn hơn 1, rồi trong quá trình gửi nhận thì thử kill 1 vài broker, rồi cho consumer đọc output để xem kết quả thế nào,…

Bước 7: xóa quyền admin của user kafka

Sau khi đã cài đặt và test xong thì bạn nên loại bỏ quyền được su của user kafka đi bằng cách:

gpasswd -d kafka wheel

Khóa password của user kafka, lúc này sẽ không có ai có thể đăng nhập trực tiếp bằng user kafka được nữa:

passwd kafka -l

Bài hướng dẫn cài đặt Kafka trên CentOS 7 đến đây là hết, cảm ơn các bạn đã theo dõi.

Trong bài tới mình sẽ hướng dẫn nhiều hơn về cách sử dụng Kafka và ứng dụng thực tế Kafka.

Thân!