Kafka消费者在重启后无法读取消息。我正在使用weiboad / kafka-php。

您可以尝试以下解决方法：

1. 使用Kafka的offset来保存消费者的位置。在重启后，您可以使用保存的offset来恢复到上次的位置，并继续消费消息。以下是一个示例代码：

$conf = new RdKafka\Conf();
$conf->set('group.id', 'my-consumer-group');
$consumer = new RdKafka\KafkaConsumer($conf);

$consumer->subscribe(['my-topic']);

// 获取保存的offset
$offset = loadOffsetFromStorage();

// 设置消费者的起始位置
$consumer->seek('my-topic', $offset, 0);

while (true) {
    $message = $consumer->consume(120 * 1000);

    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // 处理消息
            processMessage($message);

            // 保存当前消费的offset
            saveOffsetToStorage($message->offset + 1);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            // 消息已经消费完毕
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            // 没有新消息，继续下一次消费
            break;
        default:
            // 其他错误处理
            handleError($message);
            break;
    }
}

2. 检查您的消费者是否正确配置了自动提交offset。如果自动提交被禁用，您需要在处理完消息后手动提交offset。以下是一个示例代码：

$conf = new RdKafka\Conf();
$conf->set('group.id', 'my-consumer-group');
$conf->set('enable.auto.commit', 'false'); // 禁用自动提交offset
$consumer = new RdKafka\KafkaConsumer($conf);

$consumer->subscribe(['my-topic']);

while (true) {
    $message = $consumer->consume(120 * 1000);

    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            // 处理消息
            processMessage($message);

            // 手动提交offset
            $consumer->commit($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            // 消息已经消费完毕
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            // 没有新消息，继续下一次消费
            break;
        default:
            // 其他错误处理
            handleError($message);
            break;
    }
}

请根据您的实际情况选择适合您的解决方法，并根据需要进行调整和修改。