ubuntu

Ubuntu Kafka消费者的配置指南

小樊
43
2025-05-20 19:01:19
栏目: 智能运维

在Ubuntu上配置Kafka消费者涉及多个步骤,包括安装必要的软件、配置Kafka和Zookeeper,以及创建和运行消费者应用程序。以下是一个详细的指南:

1. 安装Java环境

首先,确保你的系统上安装了Java运行时环境。你可以使用以下命令来安装OpenJDK 8:

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

2. 安装Zookeeper

下载并解压Zookeeper:

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

配置Zookeeper:

sudo nano /usr/local/zookeeper/conf/zoo.cfg

添加以下内容:

tickTime=2000
dataDir=/var/lib/zookeeper
clientPort=2181

启动Zookeeper:

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

验证Zookeeper是否启动成功:

sudo netstat -nap | grep 2181

3. 安装Kafka

下载并解压Kafka:

wget https://archive.apache.org/dist/kafka/2.1.0/kafka_2.11-2.1.0.tgz
tar -xzvf kafka_2.11-2.1.0.tgz
sudo mv kafka_2.11-2.1.0 /opt/kafka

配置Kafka:

sudo nano /opt/kafka/config/server.properties

确保以下配置正确:

broker.id=0
listeners=PLAINTEXT://:9092
log.dirs=/opt/kafka/data
zookeeper.connect=localhost:2181
delete.topic.enable=true

启动Kafka服务:

sudo /opt/kafka/bin/zookeeper-server-start.sh ../config/zookeeper.properties
sudo /opt/kafka/bin/kafka-server-start.sh ../config/server.properties

验证Kafka是否启动成功:

sudo netstat -an | grep 9092

4. 创建和测试Topic

创建一个测试Topic:

sudo /opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

验证Topic是否创建成功:

sudo /opt/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181

5. 配置消费者组

在消费者应用程序中,你需要指定消费者组ID。以下是一个使用Kafka-Python客户端的示例:

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test',
    bootstrap_servers='localhost:9092',
    group_id='my-consumer-group'
)

for msg in consumer:
    print(msg.value)

6. 运行消费者组

运行你的消费者应用程序,确保它能够加入消费者组并消费消息。

7. 使用Spring Boot配置Kafka消费者

如果你使用Spring Boot,可以参考以下示例代码:

@Component
@Slf4j
public class MessageReceiveListener {

    @KafkaListener(topics = "ifaas-test", containerFactory = "ifaasContainerFactory")
    public void receiveMessage1(ListConsumerRecord consumerRecords, Acknowledgment ack) {
        try {
            log.info("receiveMessage1 接收的kafka消息:" + consumerRecords.size());
            ack.acknowledge();
        } catch (Exception e) {
            log.error("kafka失败消息:{}", JSON.toJSONString(consumerRecords));
        }
    }

    @Bean("ifaasContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> ifaasContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        Map<String, Object> props = staticConsumerProps("localhost:9092", "my-consumer-group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(props));
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        factory.setBatchListener(true);
        factory.setConcurrency(1);
        return factory;
    }

    private Map<String, Object> staticConsumerProps(String servers, String groupId) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 300000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 350000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 4000);
        return props;
    }
}

以上步骤应该帮助你在Ubuntu上配置Kafka消费者。如果在配置过程中遇到问题,请检查日志文件以获取更多信息。

0
看了该问题的人还看了