我準(zhǔn)備了一個(gè) kafka 生產(chǎn)者，它將一個(gè) List 放入 kafka 主題中。它適用于 100 萬(wàn)行/記錄。我得到的生產(chǎn)文件包含 1.1 億多條記錄。 在我的 KafkaProducer 處理如此龐大的數(shù)據(jù)的最佳方法是什么？下面是代碼，我曾經(jīng)處理過(guò) 100 萬(wàn)條記錄，將其放入 kafka 主題大約需要 4 分鐘。import java.io.BufferedReader;import java.io.File;import java.io.FileInputStream;import java.io.IOException;import java.io.InputStreamReader;import java.io.RandomAccessFile;import java.nio.charset.StandardCharsets;import java.util.ArrayList;import java.util.List;import java.util.Map;import org.apache.kafka.connect.data.Schema;import org.apache.kafka.connect.data.SchemaBuilder;import org.apache.kafka.connect.data.Struct;import org.apache.kafka.connect.source.SourceRecord;import org.apache.kafka.connect.source.SourceTask;public class KafkaSourceTask extends SourceTask {    private String filename;    private String topic;    private RandomAccessFile raf;    private long lastRecordedOffset = 0L;    private BufferedReader bufferedReader = null;    Schema schema = SchemaBuilder.struct().field("emp_id",             Schema.STRING_SCHEMA).field("name", Schema.STRING_SCHEMA)            .field("last_name", Schema.STRING_SCHEMA).field("department",             Schema.STRING_SCHEMA).build();public void start(Map<String, String> props) {    filename = props.get("file");    topic = props.get("topic");}對(duì)此的任何幫助或建議將不勝感激。在此先感謝您。