From e1c21b67fa927377d65f6811f3fba43286b9f3bc Mon Sep 17 00:00:00 2001 From: zhangxin10 Date: Thu, 12 Nov 2015 15:35:39 +0800 Subject: [PATCH] =?UTF-8?q?1.=20=E4=BF=AE=E6=94=B9=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E6=8E=A5=E5=8F=97=E6=95=B0=E6=8D=AE=E7=9A=84=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../skywalking/reciever/CollectionServer.java | 8 +++++--- .../reciever/buffer/DataBufferThread.java | 16 ++++++++++++---- .../cloud/skywalking/reciever/conf/Config.java | 2 +- .../reciever/persistance/PersistenceThread.java | 4 ++-- .../src/main/resources/config.properties | 6 +++--- 5 files changed, 23 insertions(+), 13 deletions(-) diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java index 47a98523a..33d709659 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/CollectionServer.java @@ -63,7 +63,8 @@ public class CollectionServer { public static void readDataFromSocketChannel(int length, Map byteBuffers, ByteChannel byteChannel) { int tmp = length; - StringBuilder stringBuilder = new StringBuilder(); + byte[] tmpBytes = new byte[length]; + int tmpLength = 0; for (Map.Entry entry : byteBuffers.entrySet()) { int j = tmp / entry.getKey(); if (j == 0) { @@ -72,16 +73,17 @@ public class CollectionServer { for (int k = 0; k < j; k++) { try { byteChannel.read(entry.getValue()); - stringBuilder.append(new String(entry.getValue().array())); + System.arraycopy(entry.getValue().array(), 0, tmpBytes, tmpLength, entry.getValue().array().length); } catch (IOException e) { logger.error("Read data From socket channel", e); } finally { entry.getValue().clear(); } + tmpLength += entry.getKey(); } tmp = tmp % entry.getKey(); } - //DataBufferThreadContainer.getDataBufferThread().doCarry(stringBuilder.toString()); + DataBufferThreadContainer.getDataBufferThread().doCarry(tmpBytes); } private void beginToRead(ServerSocketChannel serverSocketChannel, SelectionKey key) throws IOException { diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java index 27ec6f84e..ec841e2d8 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/buffer/DataBufferThread.java @@ -15,7 +15,7 @@ import static com.ai.cloud.skywalking.reciever.conf.Config.Buffer.*; public class DataBufferThread extends Thread { private Logger logger = LogManager.getLogger(DataBufferThread.class); - private String[] data = new String[PER_THREAD_MAX_BUFFER_NUMBER]; + private byte[][] data = new byte[PER_THREAD_MAX_BUFFER_NUMBER][]; private File file; private FileOutputStream outputStream; private AtomicInteger index = new AtomicInteger(); @@ -27,7 +27,7 @@ public class DataBufferThread extends Thread { file.createNewFile(); } - if (logger.isDebugEnabled()){ + if (logger.isDebugEnabled()) { logger.debug("Create buffer data file {}.", file.getName()); } outputStream = new FileOutputStream(file, true); @@ -43,6 +43,7 @@ public class DataBufferThread extends Thread { @Override public void run() { boolean isWriteFailure; + byte[] tmpByte; while (true) { boolean bool = false; int count = 0; @@ -55,7 +56,8 @@ public class DataBufferThread extends Thread { isWriteFailure = true; while (isWriteFailure) { try { - outputStream.write((data[i] + "\n").getBytes()); + outputStream.write(data[i]); + outputStream.write("\n".getBytes()); isWriteFailure = false; } catch (IOException e) { logger.error("Write buffer data failed.", e); @@ -117,7 +119,7 @@ public class DataBufferThread extends Thread { } - public void doCarry(String s) { + public void doCarry(byte[] s) { int i = Math.abs(index.getAndIncrement() % data.length); while (data[i] != null) { try { @@ -129,4 +131,10 @@ public class DataBufferThread extends Thread { data[i] = s; } + + public static void main(String[] args) { + byte[][] bytes = new byte[1024][]; + bytes[0] = new byte[]{'1', '2', '3'}; + System.out.println(bytes[0]); + } } diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java index 3399ae50f..ca1becc70 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/conf/Config.java @@ -11,7 +11,7 @@ public class Config { // 数据缓存配置类 public static class Buffer { // 最大数据缓存线程数量 - public static int MAX_THREAD_NUMBER = 1; + public static int MAX_THREAD_NUMBER = 3; //每个线程最大缓存数量 public static int PER_THREAD_MAX_BUFFER_NUMBER = 1024; diff --git a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java index 98ae8c0f0..8538ac8a3 100644 --- a/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java +++ b/skywalking-server/src/main/java/com/ai/cloud/skywalking/reciever/persistance/PersistenceThread.java @@ -35,7 +35,7 @@ public class PersistenceThread extends Thread { logger.debug("Get file[{}] offset [{}]", file1.getName(), offset); } char[] chars = new char[OFFSET_FILE_READ_BUFFER_SIZE]; - StringBuffer data = new StringBuffer(); + StringBuilder data = new StringBuilder(2048); boolean bool = true; length = 0; while (bool) { @@ -51,7 +51,7 @@ public class PersistenceThread extends Thread { continue; } // HBase - System.out.println(data); + //System.out.println(data); if ("EOF".equals(data.toString())) { bufferedReader.close(); diff --git a/skywalking-server/src/main/resources/config.properties b/skywalking-server/src/main/resources/config.properties index bf6fba490..2b1c50d64 100644 --- a/skywalking-server/src/main/resources/config.properties +++ b/skywalking-server/src/main/resources/config.properties @@ -17,11 +17,11 @@ buffer.data_file_max_length=104857600 buffer.flush_number_of_cache=30 #最大持久化的线程数量 -persistence.max_thread_number=2 +persistence.max_thread_number=3 #定位文件时,每次读取偏移量跳过大小 -persistence.offset_file_skip_length=2048 +persistence.offset_file_skip_length=20480 #每次读取文件偏移量大小 -persistence.offset_file_read_buffer_size=2048 +persistence.offset_file_read_buffer_size=20480 #处理文件完成之后，等待时间(单位:毫秒) persistence.switch_file_wait_time=5000 #追加EOF标志位的线程数量