目的

源碼:

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;
24. import kafka.javaapi.producer.Producer;
25. import kafka.producer.KeyedMessage;
26. import kafka.producer.ProducerConfig;
29. /*
30. * 自己在源服務器寫生產者往kafka插入數據，註意文件"producer.properties放在linux下該jar文件同一目錄
31. * 監聽某個目錄下的文件數據然後寫入kafka
32. * nohup java -jar portallog_producer.jar portallog /var/apache/logs portallog.position >/home/sre/portalhandler/handler.log 2>&1 &
33. *
34. *
35. */
36. public class PortalLogTail_Line {
38. private Producer<String,String> inner;
39. java.util.Random ran = new Random();
40. public PortalLogTail_Line() throws FileNotFoundException, IOException {
41. Properties properties = new Properties();
42. // properties.load(ClassLoader.getSystemResourceAsStream("producer.properties"));
44. properties.load(new FileInputStream("producer.properties"));
46. ProducerConfig config = new ProducerConfig(properties);
48. inner = new Producer<String, String>(config);
50. }
53. public void send(String topicName,String message) {
54. if(topicName == null || message == null){
55. return;
56. }
57. // KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message);
58. //隨機作為key,hash分散到各個分區
59. KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,String.valueOf(ran.nextInt(9)),message);
60. // KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,message,message);
61. inner.send(km);
63. }
65. public void send(String topicName,Collection<String> messages) {
66. if(topicName == null || messages == null){
67. return;
68. }
69. if(messages.isEmpty()){
70. return;
71. }
72. List<KeyedMessage<String, String>> kms = new ArrayList<KeyedMessage<String, String>>();
73. for(String entry : messages){
74. KeyedMessage<String, String> km = new KeyedMessage<String, String>(topicName,entry);
75. kms.add(km);
76. }
77. inner.send(kms);
78. }
80. public void close(){
81. inner.close();
82. }
86. public String getNewFile(File file)
87. {
88. File[] fs=file.listFiles();
89. long maxtime=0;
90. String newfilename="";
91. for (int i=0;i<fs.length;i++)
92. {
93. if (fs[i].lastModified()>maxtime && fs[i].getName().contains("access"))
94. {
95. maxtime=fs[i].lastModified();
96. newfilename=fs[i].getAbsolutePath();
98. }
99. }
100. return newfilename;
101. }
102. //寫入文件名及行號
103. public void writePosition(String path,int rn,String positionpath)
104. {
105. try {
106. BufferedWriter out = new BufferedWriter(new FileWriter(positionpath));
107. out.write(path+","+rn);
108. out.close();
109. } catch (IOException e) {
110. }
111. }
112. LineNumberReader randomFile=null;
113. String newfile=null;
114. String thisfile=null;
115. String prefile=null;
116. int ln=0;
117. int beginln=0;
118. public void realtimeShowLog(final File file,final String topicname, final String positionpath) throws IOException{
120. //啟動一個線程每1秒鐘讀取新增的日誌信息
121. new Thread(new Runnable(){
122. public void run() {
123. thisfile=getNewFile(file);
124. prefile=thisfile;
125. //訪問position文件，如果記錄了文件路徑，及行號，則定位，否則使用最新的文件
126. try {
127. BufferedReader br=new BufferedReader(new FileReader(positionpath));
128. String line=br.readLine();
129. if (line!=null &&line.contains(","))
130. {
131. thisfile=line.split(",")[0];
132. prefile=thisfile;
133. beginln=Integer.parseInt(line.split(",")[1]);
134. }
137. } catch (FileNotFoundException e2) {
138. // TODO Auto-generated catch block
139. e2.printStackTrace();
140. }
141. catch (IOException e2) {
142. // TODO Auto-generated catch block
143. e2.printStackTrace();
144. }
146. //指定文件可讀可寫
147. try {
148. randomFile = new LineNumberReader(new FileReader(thisfile));
149. } catch (FileNotFoundException e) {
150. // TODO Auto-generated catch block
151. e.printStackTrace();
152. }
153. while (true)
154. {
155. try {
156. Thread.sleep(100);
158. } catch (InterruptedException e1) {
159. // TODO Auto-generated catch block
160. e1.printStackTrace();
161. }
162. try {
163. //獲得變化部分的
164. // randomFile.seek(lastTimeFileSize);
165. String tmp = "";
166. while( (tmp = randomFile.readLine())!= null) {
167. int currln=randomFile.getLineNumber();
168. //beginln默認為0
169. if (currln>beginln)
170. send(topicname,new String(tmp.getBytes("utf8")));
172. ln++;
174. //每發生一條寫一次影響效率,連續發100次後再記錄位置
175. if (ln>100)
176. {
177. writePosition(thisfile,currln,positionpath);
178. ln=0;
179. }
181. }
182. thisfile=getNewFile(file);
183. if(!thisfile.equals(prefile))
185. {
186. randomFile.close();
187. randomFile = new LineNumberReader(new FileReader(thisfile));
188. prefile=thisfile;
189. beginln=0;
190. }
193. } catch (IOException e) {
194. throw new RuntimeException(e);
195. }
196. }
197. }}).start();
198. }
200. /**
201. * @param args
202. * @throws Exception
203. */
204. public static void main(String[] args) throws Exception {
205. PortalLogTail_Line producer = new PortalLogTail_Line();
206. if (args.length!=3)
207. {
208. System.out.println("usage:topicname pathname positionpath");
209. System.exit(1);
210. }
211. String topicname=args[0];
212. String pathname=args[1];
213. String positionpath=args[2];
214. final File tmpLogFile = new File(pathname);
215. producer.realtimeShowLog(tmpLogFile,topicname,positionpath);
219. }
221. }

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

測試

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

java實時監聽日誌寫入kafka