弹幕采集添加数据库存储提交

pom.xml

@@ -15,6 +15,12 @@
   		<version>0.6.2.1-RELEASE</version>
   	</dependency>
   	
+  	<dependency>
+        <groupId>org.mongodb</groupId>
+        <artifactId>mongo-java-driver</artifactId>
+        <version>3.11.0</version>
+    </dependency>
+  	
   	<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
 		<dependency>
 			<groupId>org.apache.commons</groupId>

src/main/java/com/zhiwei/live/danmu/bilibili/BilibiliMessageHandler.java

@@ -89,8 +89,6 @@ public class BilibiliMessageHandler extends ChannelInboundHandlerAdapter{
      * @param data
      * @throws Exception
      */
-
-
     private void analyzeData(byte[] data) throws IOException, DataFormatException {
         int dataLength = data.length;
         if (dataLength < 16) {
@@ -110,7 +108,9 @@ public class BilibiliMessageHandler extends ChannelInboundHandlerAdapter{
                 if (action == 2) {
                     inputStream.readInt();
                     int userCount = inputStream.readInt();
-                    System.out.println(userCount);
+                    JSONObject json = new JSONObject();
+                    json.put("userCount", userCount);
+                    dataCallBack.onData(json);
                 } else if (action == 4) {
                     int param = inputStream.readInt();
                     int msgBodyLength = dataLength - 16;
@@ -119,7 +119,7 @@ public class BilibiliMessageHandler extends ChannelInboundHandlerAdapter{
                     if (version != 2) {
                         String jsonStr = new String(msgBody, StandardCharsets.UTF_8);
                         JSONObject dataJson = JSONObject.parseObject(jsonStr);
-                        System.out.println(dataJson);
+                        dataCallBack.onData(dataJson);
                     } else {
                         Inflater inflater = new Inflater();
                         inflater.setInput(msgBody);
@@ -137,7 +137,7 @@ public class BilibiliMessageHandler extends ChannelInboundHandlerAdapter{
                             if (innerAction == 4) {
                                 String jsonStr = new String(innerData, StandardCharsets.UTF_8);
                                 JSONObject dataJson = JSONObject.parseObject(jsonStr);
-                                System.out.println(dataJson);
+                                dataCallBack.onData(dataJson);
                             } else if (innerAction == 2) {
                                 // pass
                             }

src/main/java/com/zhiwei/live/danmu/util/DataCallBack.java

 package com.zhiwei.live.danmu.util;
 
-import com.zhiwei.live.danmu.bilibili.BilibiliMessage;
-import com.zhiwei.live.danmu.douyu.DouYuMessage;
-import com.zhiwei.live.danmu.pandam.PandamMessage;
-
 public interface DataCallBack {
     
     /**
@@ -15,8 +11,4 @@ public interface DataCallBack {
      */
     void onData(Object message);
     
-    
-    
-    
-    
 }

src/test/java/com/zhiwei/live/test/danmu/BilibiliDanMuTest.java

 package com.zhiwei.live.test.danmu;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.bson.Document;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DB;
+import com.mongodb.DBCollection;
+import com.mongodb.DBObject;
+import com.mongodb.Mongo;
 import com.zhiwei.live.danmu.bilibili.BilibiliClient;
-import com.zhiwei.live.danmu.bilibili.BilibiliMessage;
 import com.zhiwei.live.danmu.util.DataCallBack;
+import com.zhiwei.tools.tools.ZhiWeiTools;
 
 public class BilibiliDanMuTest {
 	
+    private static BlockingQueue<Object> dataQuery = new ArrayBlockingQueue<>(100000);
+    
 	public static void main(String[] args) throws InterruptedException {
-    	String roomUrl = "https://live.bilibili.com/387";
+	    Mongo mongo = new Mongo("127.0.0.1", 27017);
+	    DB mdb = mongo.getDB("bilibili");
+	    DBCollection dbc = mdb.getCollection("bilibili_danmu");
+    	String roomUrl = "https://live.bilibili.com/650";
     	try {
 			BilibiliClient.getDanmu(new DataCallBack() {
                 @Override
                 public void onData(Object message) {
-                	if(message instanceof BilibiliMessage) {
-                		BilibiliMessage bilibiliMessage = (BilibiliMessage)message;
-                		 System.out.println("-------------" + bilibiliMessage.toString());
-                	}
+                    System.out.println(message);
+                    dataQuery.add(message);
                 }
             }, roomUrl);
+			Long time = System.currentTimeMillis();
+			while(true) {
+			    try {
+			        if(dataQuery.size() > 20 || System.currentTimeMillis() - time > 5000) {
+			            List<DBObject> docList = new ArrayList<>();
+			            while(true) {
+			                Object message = dataQuery.poll();
+			                if(Objects.nonNull(message)) {
+			                    docList.add(new BasicDBObject("data", String.valueOf(message)));
+			                }else {
+			                    break;
+			                }
+			            }
+			            System.out.println("此轮存储量 " + docList.size());
+			            if(!docList.isEmpty()) {
+			                dbc.insert(docList);
+			            }
+			            time = System.currentTimeMillis();
+			        }
+                } catch (Exception e) {
+                }
+			}
 		} catch (Exception e) {
 			e.printStackTrace();
 		}