如何根據業務需求配置Ubuntu Kafka

根據業務需求配置Ubuntu Kafka涉及多個步驟，包括安裝和配置Kafka、Zookeeper，以及根據具體業務場景調整Kafka的配置參數。以下是一個詳細的指南：

1. 安裝Java和Zookeeper

首先，確保你的Ubuntu系統上安裝了Java和Zookeeper。

安裝Java

sudo apt update
sudo apt install openjdk-8-jdk
java -version

安裝Zookeeper

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz
tar xvf zookeeper-3.4.6.tar.gz
sudo mv zookeeper-3.4.6 /usr/local/zookeeper

配置Zookeeper：

sudo cat > /usr/local/zookeeper/conf/zoo.cfg << EOF
tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181
EOF

啟動Zookeeper：

sudo /usr/local/zookeeper/bin/zkServer.sh start

2. 安裝Kafka

下載并解壓Kafka：

wget https://downloads.apache.org/kafka/3.5.2/kafka_2.12-3.5.2.tgz
tar -zvxf kafka_2.12-3.5.2.tgz
sudo mv kafka_2.12-3.5.2 /usr/local/kafka

配置Kafka： 編輯/usr/local/kafka/config/server.properties文件，設置以下參數：

• broker.id: 每個Kafka Broker的唯一標識符。
• listeners: Kafka Broker監聽的網絡地址和端口。
• log.dirs: Kafka Broker用于存儲消息日志的目錄。

示例配置：

broker.id=1
listeners=PLAINTEXT://your_server_ip:9092
log.dirs=/tmp/kafka-logs

啟動Kafka：

sudo /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties

3. 創建Topic

創建一個Topic，用于生產者和消費者之間的通信：

sudo /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic your_topic_name

4. 配置Kafka生產者

根據業務需求，可以配置Kafka生產者以發送消息到Kafka集群。以下是一個簡單的生產者示例：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "your_server_ip:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.send(new ProducerRecord<>("your_topic_name", "key", "value"));
        producer.close();
    }
}

5. 配置Kafka消費者

根據業務需求，可以配置Kafka消費者以從Kafka集群接收消息。以下是一個簡單的消費者示例：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "your_server_ip:9092");
        props.put("group.id", "your_group_id");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("your_topic_name"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

6. 根據業務需求調整配置

根據具體的業務需求，可能需要調整以下配置參數：

• acks: 生產者確認消息已經被寫入Leader的參數，可以是0（不等待確認）、1（等待Leader寫入Leader）或all（等待Leader寫入所有ISR）。
• retries: 生產者在遇到可恢復的錯誤時重試的次數。
• batch.size: 生產者批量發送消息的大小。
• linger.ms: 生產者在發送消息前等待更多消息加入批次的時間。
• buffer.memory: 生產者可以使用的最大內存量。

7. 開機自啟配置

為了確保Kafka在系統重啟后自動啟動，可以配置開機自啟：

sudo systemctl enable zookeeper
sudo systemctl enable kafka
sudo systemctl start zookeeper
sudo systemctl start kafka