kafkaProducer 1.1 读取文件目录文件,发送到kafka Topic
时间:2018-12-20
本文章向大家介绍kafkaProducer 1.1 读取文件目录文件,发送到kafka Topic,主要包括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();
}
}
}
- JavaScript 教程
- JavaScript 编辑工具
- JavaScript 与HTML
- JavaScript 与Java
- JavaScript 数据结构
- JavaScript 基本数据类型
- JavaScript 特殊数据类型
- JavaScript 运算符
- JavaScript typeof 运算符
- JavaScript 表达式
- JavaScript 类型转换
- JavaScript 基本语法
- JavaScript 注释
- Javascript 基本处理流程
- Javascript 选择结构
- Javascript if 语句
- Javascript if 语句的嵌套
- Javascript switch 语句
- Javascript 循环结构
- Javascript 循环结构实例
- Javascript 跳转语句
- Javascript 控制语句总结
- Javascript 函数介绍
- Javascript 函数的定义
- Javascript 函数调用
- Javascript 几种特殊的函数
- JavaScript 内置函数简介
- Javascript eval() 函数
- Javascript isFinite() 函数
- Javascript isNaN() 函数
- parseInt() 与 parseFloat()
- escape() 与 unescape()
- Javascript 字符串介绍
- Javascript length属性
- javascript 字符串函数
- Javascript 日期对象简介
- Javascript 日期对象用途
- Date 对象属性和方法
- Javascript 数组是什么
- Javascript 创建数组
- Javascript 数组赋值与取值
- Javascript 数组属性和方法
- 制作的 dotnet tool 运行失败提示依赖缺失
- 撸了个多线程断点续传下载器,我从中学习到了这些知识
- 使用 IOC 控制反转和 DI 依赖注入的意义
- Servlet配置初始值,并获取初始值
- CSS 技巧一则 -- 不定宽溢出文本适配滚动
- 潘石屹用Python解决100个问题 | 最大公约数
- 我的天上传文件又出现问题了(超出大小限制)
- 现在学 PHP 没有发展?来看看这个后台框架你还会这么想吗
- leetcode-easy-array-最大子序和
- Leecode No.3 无重复字符的最长子串
- Solidity 0.6.11 更新
- 潘石屹用Python解决100个问题 | 最小公倍数
- 通过CREATE2获得合约地址:解决交易所充值账号问题
- 小知识:如何判定crontab任务的执行频度
- 以太坊合约静态分析工具Slither简介与使用