在Ubuntu上配置Kafka消费者涉及多个步骤，包括安装必要的软件、配置Kafka和Zookeeper，以及创建和运行消费者应用程序。以下是一个详细的指南：
首先，确保你的系统上安装了Java运行时环境。你可以使用以下命令来安装OpenJDK 8：
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 -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 下载并解压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 创建一个测试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 在消费者应用程序中，你需要指定消费者组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) 运行你的消费者应用程序，确保它能够加入消费者组并消费消息。
如果你使用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消费者。如果在配置过程中遇到问题，请检查日志文件以获取更多信息。
