Kafka是一個分布式的消息系統，它可以處理海量的消息數據。在Kafka中，生產者將消息寫入到一個或多個主題(topic)中，消費者從某個主題中讀取消息并進行處理。

消費者在處理消息時需要控制消息的順序和重復消費的問題，這時就需要使用Kafka中的偏移量(offset)來記錄消費者在某個主題中已經讀取到的消息。

偏移量的保存方式有很多種，比如可以保存在ZooKeeper中或者保存在本地文件中。在實際應用中，我們可以將偏移量保存在MySQL中。

public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
for (TopicPartition partition : partitions) {
OffsetAndMetadata offsetAndMetadata = loadOffsetFromDB(partition);
if (offsetAndMetadata != null) {
consumer.seek(partition, offsetAndMetadata.offset());
}
}
}
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
for (TopicPartition partition : partitions) {
commitOffsetToDB(partition, consumer.committed(partition).offset());
}
}
private OffsetAndMetadata loadOffsetFromDB(TopicPartition partition) {
// 從MySQL中加載offset
}
private void commitOffsetToDB(TopicPartition partition, long offset) {
// 將offset保存到MySQL中
}

在消費者的代碼中，我們可以使用類似上面的代碼來實現將偏移量保存到MySQL中。在onPartitionsAssigned方法中，我們從MySQL中讀取消費者在該分區中的偏移量；在onPartitionsRevoked方法中，我們將消費者在該分區中的偏移量提交到MySQL中。這樣就可以確保消費者在下次啟動時可以從上一次消費的位置繼續消費。

當然，在保存偏移量的時候，我們也需要考慮數據庫的容災、性能等問題，需要根據具體情況來選擇最合適的方案。