1.java实现读取本地数据到kafka生产数据
/** * Created by 王一宁 on 2019/11/6. */ public class kafkaProducer { public static void main(String[] args) throws Exception{ Properties prop = new Properties(); //指定kafka broker地址 prop.put("bootstrap.servers", "hadoop1:9092"); //指定key value的序列化方式 prop.put("key.serializer", StringSerializer.class.getName()); prop.put("value.serializer", StringSerializer.class.getName()); //指定topic名称 String topic = "wang"; //创建producer链接 KafkaProducer<String, String> producer = new KafkaProducer<String,String>(prop); //创建Java IO InputStream file = new FileInputStream("D:\\APP\\IDEA\\workplace\\FlinkTurbineFaultDiagnosis\\src\\main\\resources\\turbine\\GW20000120160101.txt"); InputStreamReader fileInputStream = new InputStreamReader(file); BufferedReader reader = new BufferedReader(fileInputStream); String line = null; while ((line = reader.readLine()) != null) { //生产消息 producer.send(new ProducerRecord<String, String>(topic,line)); Thread.sleep(1000); } reader.close(); file.close(); fileInputStream.close(); //关闭链接 producer.close(); } }
2.在linux服务器中，直接开启一个消费者，就可以看到生产的数据了，或者手写一个java消费者，消费同一个Topic的数据。
3.java实现flink集成kafka消费者的实现代码
/** * 消费Kafka中得数据 * @author 王一宁 * @date 2020/1/2 12:12 */ public class StreamingFromKafka { public static void main(String[] args) throws Exception{ //获取环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); //kafka配置 String topic = "wang"; Properties prop = new Properties(); prop.setProperty("bootstrap.servers","hadoop1:9092");//多个的话可以指定 prop.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); prop.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); prop.setProperty("auto.offset.reset","latest"); prop.setProperty("group.id","consumer1"); FlinkKafkaConsumer010<String> myConsumer = new FlinkKafkaConsumer010<String>(topic, new SimpleStringSchema(), prop); //获取数据 DataStream<String> text = env.addSource(myConsumer); //打印 text.print().setParallelism(1); //执行 //env.execute("StreamingFormCollection"); env.execute(); } }
