1. 修改服务接受数据的方式

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<Integer, ByteBuffer> byteBuffers, ByteChannel byteChannel) {
         int tmp = length;
-        StringBuilder stringBuilder = new StringBuilder();
+        byte[] tmpBytes = new byte[length];
+        int tmpLength = 0;
         for (Map.Entry<Integer, ByteBuffer> 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 {

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]);
+    }
 }

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;

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();

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标志位的线程数量