大数据纠错笔记——数据写入kafka

这里选择用JAVA代码实现将数据写入kafka：

package work;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.HashMap;
import java.util.Map;

public class Test_kafka {
    public static void main(String[] args) throws Exception {
        BufferedReader br = new BufferedReader(new FileReader("C:\Users\a.txt"));

        Map map = new HashMap();
        map.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop10:9092");
        map.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        map.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        KafkaProducer producer = new KafkaProducer(map);
//        producer.send(new ProducerRecord<>("topica","aaaaa"));
        String line = null;
        while((line = br.readLine()) != null){
            if (StringUtils.isNotBlank(line.trim())){
//                System.out.println(line);
                producer.send(new ProducerRecord<>("topica",line));
            }
        }

        producer.close(); // ！这里必须写，否则会报错
        br.close();
    }
}

运行后出现如上错误：

首先producer.close();代码必须写

第二可能是因为导包导错，需要注意