java实时监听日志写入kafka

目的

实时监听某目录下的日志文件,如有新文件切换到新文件,并同步写入kafka,同时记录日志文件的行位置,以应对进程异常退出,能从上次的文件位置开始读取(考虑到效率,这里是每100条记一次,可调整)

源码:

[java] view plain copy

  1. import java.io.BufferedReader;
  2. import java.io.BufferedWriter;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileNotFoundException;
  6. import java.io.FileReader;
  7. import java.io.FileWriter;
  8. import java.io.IOException;
  9. import java.io.LineNumberReader;
  10. import java.io.PrintWriter;
  11. import java.io.RandomAccessFile;
  12. import java.net.NoRouteToHostException;
  13. import java.util.ArrayList;
  14. import java.util.Collection;
  15. import java.util.List;
  16. import java.util.Properties;
  17. import java.util.Random;
  18. import java.util.concurrent.Executors;
  19. import java.util.concurrent.ScheduledExecutorService;
  20. import java.util.concurrent.TimeUnit;
  21. import kafka.javaapi.producer.Producer;
  22. import kafka.producer.KeyedMessage;
  23. import kafka.producer.ProducerConfig;
  24. /*
  25. * 自己在源服务器写生产者往kafka插入数据,注意文件"producer.properties放在linux下该jar文件同一目录
  26. * 监听某个目录下的文件数据然后写入kafka
  27. * nohup java -jar portallog_producer.jar portallog /var/apache/logs portallog.position  >/home/sre/portalhandler/handler.log 2>&1 &
  28. *
  29. *
  30. */
  31. public class PortalLogTail_Line {
  32. private Producer<String,String> inner;
  33. java.util.Random ran = new Random();
  34. public PortalLogTail_Line() throws FileNotFoundException, IOException {
  35. Properties properties = new Properties();
  36. //   properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));
  37. properties.load(new FileInputStream("producer.properties"));
  38. ProducerConfig config = new ProducerConfig(properties);
  39. inner = new Producer<String, String>(config);
  40. }
  41. public void send(String topicName,String message) {
  42. if(topicName == null || message == null){
  43. return;
  44. }
  45. //   KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message);
  46. //随机作为key,hash分散到各个分区
  47. KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,String.valueOf(ran.nextInt(9)),message);
  48. //   KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message,message);
  49. inner.send(km);
  50. }
  51. public void send(String topicName,Collection<String> messages) {
  52. if(topicName == null || messages == null){
  53. return;
  54. }
  55. if(messages.isEmpty()){
  56. return;
  57. }
  58. List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>();
  59. for(String entry : messages){
  60. KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,entry);
  61. kms.add(km);
  62. }
  63. inner.send(kms);
  64. }
  65. public void close(){
  66. inner.close();
  67. }
  68. public String getNewFile(File file)
  69. {
  70. File[] fs=file.listFiles();
  71. long maxtime=0;
  72. String newfilename="";
  73. for (int i=0;i<fs.length;i++)
  74. {
  75. if (fs[i].lastModified()>maxtime && fs[i].getName().contains("access"))
  76. {
  77. maxtime=fs[i].lastModified();
  78. newfilename=fs[i].getAbsolutePath();
  79. }
  80. }
  81. return newfilename;
  82. }
  83. //写入文件名及行号
  84. public void writePosition(String path,int rn,String positionpath)
  85. {
  86. try {
  87. BufferedWriter out = new BufferedWriter(new FileWriter(positionpath));
  88. out.write(path+","+rn);
  89. out.close();
  90. } catch (IOException e) {
  91. }
  92. }
  93. LineNumberReader randomFile=null;
  94. String newfile=null;
  95. String thisfile=null;
  96. String prefile=null;
  97. int ln=0;
  98. int beginln=0;
  99. public void realtimeShowLog(final File file,final String topicname, final String positionpath) throws IOException{
  100. //启动一个线程每1秒钟读取新增的日志信息
  101. new Thread(new Runnable(){
  102. public void run() {
  103. thisfile=getNewFile(file);
  104. prefile=thisfile;
  105. //访问position文件,如果记录了文件路径,及行号,则定位,否则使用最新的文件
  106. try {
  107. BufferedReader br=new BufferedReader(new FileReader(positionpath));
  108. String line=br.readLine();
  109. if (line!=null &&line.contains(","))
  110. {
  111. thisfile=line.split(",")[0];
  112. prefile=thisfile;
  113. beginln=Integer.parseInt(line.split(",")[1]);
  114. }
  115. } catch (FileNotFoundException e2) {
  116. // TODO Auto-generated catch block
  117. e2.printStackTrace();
  118. }
  119. catch (IOException e2) {
  120. // TODO Auto-generated catch block
  121. e2.printStackTrace();
  122. }
  123. //指定文件可读可写
  124. try {
  125. randomFile = new LineNumberReader(new FileReader(thisfile));
  126. } catch (FileNotFoundException e) {
  127. // TODO Auto-generated catch block
  128. e.printStackTrace();
  129. }
  130. while (true)
  131. {
  132. try {
  133. Thread.sleep(100);
  134. } catch (InterruptedException e1) {
  135. // TODO Auto-generated catch block
  136. e1.printStackTrace();
  137. }
  138. try {
  139. //获得变化部分的
  140. //  randomFile.seek(lastTimeFileSize);
  141. String tmp = "";
  142. while( (tmp = randomFile.readLine())!= null) {
  143. int currln=randomFile.getLineNumber();
  144. //beginln默认为0
  145. if (currln>beginln)
  146. send(topicname,new String(tmp.getBytes("utf8")));
  147. ln++;
  148. //每发生一条写一次影响效率,连续发100次后再记录位置
  149. if (ln>100)
  150. {
  151. writePosition(thisfile,currln,positionpath);
  152. ln=0;
  153. }
  154. }
  155. thisfile=getNewFile(file);
  156. if(!thisfile.equals(prefile))
  157. {
  158. randomFile.close();
  159. randomFile = new LineNumberReader(new FileReader(thisfile));
  160. prefile=thisfile;
  161. beginln=0;
  162. }
  163. } catch (IOException e) {
  164. throw new RuntimeException(e);
  165. }
  166. }
  167. }}).start();
  168. }
  169. /**
  170. * @param args
  171. * @throws Exception
  172. */
  173. public static void main(String[] args) throws Exception {
  174. PortalLogTail_Line producer = new PortalLogTail_Line();
  175. if (args.length!=3)
  176. {
  177. System.out.println("usage:topicname pathname positionpath");
  178. System.exit(1);
  179. }
  180. String topicname=args[0];
  181. String pathname=args[1];
  182. String positionpath=args[2];
  183. final File tmpLogFile = new File(pathname);
  184. producer.realtimeShowLog(tmpLogFile,topicname,positionpath);
  185. }
  186. }

producer.properties文件放在同级目录下

[html] view plain copy

  1. metadata.broker.list=xxx:10909,xxx:10909
  2. # name of the partitioner class for partitioning events; default partition spreads data randomly
  3. #partitioner.class=
  4. # specifies whether the messages are sent asynchronously (async) or synchronously (sync)
  5. producer.type=sync
  6. #producer.type=async
  7. # specify the compression codec for all data generated: none , gzip, snappy.
  8. # the old config values work as well: 0, 1, 2 for none, gzip, snappy, respectivally
  9. compression.codec=none
  10. #compression.codec=gzip
  11. # message encoder
  12. serializer.class=kafka.serializer.StringEncoder

测试

最后执行:

[java] view plain copy

  1. nohup java -jar portallog_producer.jar portallog /var/apache/logs portallog.position  >/home/sre/portalhandler/handler.log 2>&1 &
  2. 转:http://blog.csdn.net/u011750989/article/details/21237251
时间: 2024-10-21 00:20:36

java实时监听日志写入kafka的相关文章

java实时监听日志写入kafka(转)

原文链接:http://www.sjsjw.com/kf_cloud/article/020376ABA013802.asp 目的 实时监听某目录下的日志文件,如有新文件切换到新文件,并同步写入kafka,同时记录日志文件的行位置,以应对进程异常退出,能从上次的文件位置开始读取(考虑到效率,这里是每100条记一次,可调整) 源码: import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.File;

java实时监听日志写入kafka(多目录)

目的 实时监听多个目录下的日志文件,如有新文件切换到新文件,并同步写入kafka,同时记录日志文件的行位置,以应对进程异常退出,能从上次的文件位置开始读取(考虑到效率,这里是每100条记一次,可调整) 源码 [java] view plain copy import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; import j

利用Maxwell组件实时监听Mysql的binlog日志

一:在linux环境下安装部署好mysql 开启binlog Vim  /etc/my.cnf mysql的binlog格式有3种,为了把binlog解析成json数据格式,要设置binlog的格式为row(binlog有三种格式:Statement.Row以及Mixed) 3重启msyql服务 service mysqld restart 查看是否已经开启binlog 查看是否已经开启binlog 是否改为row模式 /var/lib/mysql下看到生成了相应的binlog监听日志文件,如图

ORACLE清理、截断监听日志文件(listener.log)

在ORACLE数据库中,如果不对监听日志文件(listener.log)进行截断,那么监听日志文件(listener.log)会变得越来越大,想必不少人听说过关于"LISTENER.LOG日志大小不能超过2GB,超过会导致LISTENER监听器无法处理新的连接",当然这个不是真理,不会绝对出现,只是发生在老旧的32bit Linux或Unix系统下面,真实的原因是一些32bit OS自带的文件系统不支持2GB以上的文件,导致监听服务进程(tnslsnr)append write日志文件

sql server 警报管理,实时监听数据库动向,运筹帷幄之中

原文:sql server 警报管理,实时监听数据库动向,运筹帷幄之中 工作这么多年了,无论是身边的同学还是同事,发现只要搞程序员的都有一个通病---懒.懒到谁都不愿意加班,尤其是"义务"加班.即使大家都不愿意加班,但是很多时候项目赶着上线或者上线之后出错啊什么的,总得有人看着,这时候就诞生了一种新的工作制度,叫做7*24.顾名思义就是这种岗位实时都得有人看着,这确实是一件让人头疼的事情.虽然说在项目刚上线不可避免的得有7*24,但是我们可以尽量减少7*24的工作量(ps:因为7*24

Rsync+Inotify实时监听备份

说明,下面的inotify是建立在rsync的配置过程 大前提是rsync daemon 配置成功,rsync配置看上一遍博文,在客户端可以推拉数据,然后才能配置inotify服务----inotify是在客户端安装,监听需要备份的目录,然后推送到服务端 查看当前系统是否支持inotify [[email protected] bier]# uname -r 2.6.32-431.el6.i686 [[email protected] bier]# ls -l /proc/sys/fs/inot

有关监听日志的清理问题

近日,有开发人员向我反馈:“代码时有连不上数据库的情况发生”.在了解了一些基本信息之后,希望能通过查看监听日志获取问题线索.首先是通过如下方式确定监听日志的存放路径: LSNRCTL for Linux: Version 10.2.0.4.0 - Production on 21-JUN-2016 21:19:56 Copyright (c) 1991, 2007, Oracle. All rights reserved. Connecting to (ADDRESS=(PROTOCOL=tcp

移动端用js与jquery实时监听输入框值的改动

背景: 在一次移动端H5开发中,需要监听输入框值的实时变动. onchange事件肯定抛弃,因为只能失去焦点才触发. 而keyPress在Android可以触发,iOS不可以. 又不想用Android和iOS都可以触发的keyDown和keyUp. 于是,百度出了新东西:oninput![需要配合propertychange,兼容 IE9 以下版本] 用法: JS: if(isIE) { document.getElementById("input").onpropertychange

我所不知道的的监听日志问题

今天,有开发人员向我反馈:“代码时有连不上数据库的情况发生”.在了解了一些基本情况之后,希望能通过查看监听日志获取问题线索.首先是通过如下方式确定监听日志的存放路径: LSNRCTL for Linux: Version 10.2.0.4.0 - Production on 21-JUN-2016 21:19:56 Copyright (c) 1991, 2007, Oracle. All rights reserved. Connecting to (ADDRESS=(PROTOCOL=tcp