kafkaProducer 1.1 读取文件目录文件，发送到kafka Topic

package domain;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.io.*;
import java.util.Properties;
public class run {
//读取文件流以静态变量读取
static FileInputStream fis = null;
static InputStreamReader isr = null;
static BufferedReader br = null;
static String tempString = null;
static int count = 0;
//kafka Producer 静态变量创建对象
static Producer<String, String> producer = null;
static File file = null;

public static void main(String[] args) throws IOException, InterruptedException {
//读取文件形成一个持续写入kafka Producer的流
    while (true) {
        Properties props = new Properties();
        //kafka brokerlist
        props.put(\"bootstrap.servers\", \"10.62.200.200:9092,10.62.200.201:9092,10.62.200.202:9092\");
        //ack \"0,1,-1 all\"四种，1为文件patition leader完成写入就算完成
        props.put(\"acks\", \"1\");
        props.put(\"retries\", 0);
        props.put(\"batch.size\", 16384);
        props.put(\"linger.ms\", 1);
        props.put(\"buffer.memory\", 33554432);
        //必须设置(k,v)的序列化  详情见kafkaProducer 的构造函数
        props.put(\"key.serializer\", \"org.apache.kafka.common.serialization.StringSerializer\");
        props.put(\"value.serializer\", \"org.apache.kafka.common.serialization.StringSerializer\");
        producer = new KafkaProducer<>(props);
        //指定目录
        file = new File(\"/udata/DataSource\");
        File[] files = file.listFiles();
        for (File filenew : files) {

            fis = new FileInputStream(filenew);
            isr = new InputStreamReader(fis);
            br = new BufferedReader(isr);

            while ((tempString = br.readLine()) != null) {
                count++;
                //每条消息间隔2秒
                Thread.sleep(2000);

                producer.send(
                        new ProducerRecord<>(\"tmp\", Integer.toString(count), tempString));
                        //发送完成后打印发送数据
                System.out.println(tempString);
            }

        }
        producer.close();
    }
}

}