以下是使用Java编写的Apache Kafka消费者示例代码，用于从消息队列的开头可靠地读取数据：

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

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

public class KafkaConsumerExample {
    private static final String TOPIC_NAME = "your-topic-name";
    private static final String BOOTSTRAP_SERVERS = "your-bootstrap-servers";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "your-consumer-group-id");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        KafkaConsumer consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC_NAME));

        while (true) {
            ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord record : records) {
                // 在这里处理接收到的消息
                System.out.printf("Received message: key = %s, value = %s%n", record.key(), record.value());
            }
        }
    }
}

请注意，您需要将"your-topic-name"替换为要消费的实际主题的名称，将"your-bootstrap-servers"替换为Kafka集群的引导服务器地址，并将"your-consumer-group-id"替换为消费者组的唯一ID。

这段代码创建了一个Kafka消费者，并订阅了指定的主题。然后，它会以轮询的方式从Kafka集群中拉取消息，并在接收到消息时进行处理。消费者使用了StringDeserializer来反序列化键和值。在这个例子中，我们只是简单地将接收到的消息打印出来，您可以根据自己的需求进行进一步的处理。