1

demo-dev-kafka/.gitignore

+/target/
+/.settings/
+/.classpath
+/.project

demo-dev-kafka/pom.xml

+<project xmlns="http://maven.apache.org/POM/4.0.0"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<parent>
+		<groupId>net.wicp.tams</groupId>
+		<artifactId>duckula3-dev-demo</artifactId>
+		<version>0.0.1</version>
+	</parent>
+	<artifactId>demo-dev-kafka</artifactId>
+	<dependencies>
+		<dependency>
+			<groupId>net.wicp.tams</groupId>
+			<artifactId>common-binlog-alone-api</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka-clients</artifactId>
+			<version>1.0.1</version>
+		</dependency>
+	</dependencies>
+</project>
\ No newline at end of file

demo-dev-kafka/src/main/java/net/wicp/tams/duckula/demo/kafka/KafkaUtil.java

+package net.wicp.tams.duckula.demo.kafka;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import net.wicp.tams.common.binlog.alone.DuckulaAssit;
+import net.wicp.tams.common.binlog.alone.ListenerConf;
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * @author Li Changwei
+ * @version 2020-11-18 15:01
+ */
+public class KafkaUtil {
+
+    public static void main(String[] args) throws InvalidProtocolBufferException {
+            Properties properties = new Properties();
+            properties.put("bootstrap.servers", "test-kafka-001.taimei.com:9092");
+            properties.put("group.id", "table_data_collector");
+            properties.put("key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class.getName());
+            properties.put("value.deserializer", org.apache.kafka.common.serialization.ByteArrayDeserializer.class.getName());
+            properties.put("enable.auto.commit", "false");
+
+            final KafkaConsumer<String, byte[]> kafkaConsumer = new KafkaConsumer<>(properties);
+            kafkaConsumer.subscribe(Collections.singletonList("gvp-base"));
+
+            int count = 10;
+            while (count < 20) {
+                final ConsumerRecords<String, byte[]> records = kafkaConsumer.poll(2000);
+                System.out.println("轮训开始" + System.currentTimeMillis());
+                for (ConsumerRecord<String, byte[]> record : records) {
+                    final ListenerConf.DuckulaEvent parse = DuckulaAssit.parse(record.value());
+                    final List<ListenerConf.DuckulaEventItem> itemsList = parse.getItemsList();
+                    System.out.println(record);
+                    count++;
+
+                    kafkaConsumer.commitSync();
+                }
+            }
+
+            kafkaConsumer.wakeup();
+            kafkaConsumer.close();
+    }
+
+}

duckula3-dev-demo-cuslistener/pom.xml

@@ -12,7 +12,6 @@
 		<dependency>
 			<groupId>net.wicp.tams</groupId>
 			<artifactId>common-binlog-alone</artifactId>
-			<version>3.6.13</version>
 		</dependency>
 		<dependency>
 			<groupId>ch.qos.logback</groupId>

duckula3-dev-demo-cuslistener/src/main/java/aa/Test.java

+package aa;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+
+import org.apache.commons.io.FileUtils;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+
+public class Test {
+	public static void main(String[] args) throws IOException {
+		String string = FileUtils.readFileToString(new File(
+				"D:\\source\\zjh\\duckula3-dev-demo\\duckula3-dev-demo-cuslistener\\src\\main\\resources\\quea.json"));
+		JSONArray data = JSON.parseObject(string).getJSONArray("Data");
+		AtomicInteger i = new AtomicInteger(0);
+		data.forEach(o -> {
+			JSONObject o1 = (JSONObject) o;
+			String content = o1.getString("Content");
+			List<String> collect = o1.getJSONArray("ListQuestionAnswerList").stream().map(o2 -> ((JSONObject) o2))
+					.filter(o22 -> o22.getInteger("IsAnswer") == 1)
+					.map(jsonObject -> jsonObject.getString("AnswerChoice")).collect(Collectors.toList());
+			i.getAndIncrement();
+			System.out.println(i + "." + content + " :" + String.join(",", collect));
+		});
+
+	}
+}
\ No newline at end of file

duckula3-dev-demo-cuslistener/src/main/java/net/wicp/tams/duckula/demo/cuslistener/TestListener.java

@@ -11,7 +11,7 @@ import net.wicp.tams.common.binlog.alone.binlog.listener.AbsBinlogListener;
 
 public class TestListener extends AbsBinlogListener {
 	@Override
-	public void doBusiTrue(Rule rule, DuckulaEvent duckulaEvent) {
+	public void doBusiTrue(Rule rule, DuckulaEvent duckulaEvent, boolean isSplit) {
 		Integer age = DuckulaAssit.getValue(duckulaEvent, "age", 0);
 		System.out.println("age=" + age);
 		Serializable postNo = DuckulaAssit.getValue(duckulaEvent, "postNo", 0);

duckula3-dev-demo-cuslistener/src/main/resources/quea.json

+{"Message":null,"Code":"200","Data":[{"SubjectiveAnswer":null,"CompanyId":"6edc1973f7a24c65a305a6db001f52d0","ListQuestionAnswerList":[{"AnswerChoice":".csv","SubjectiveAnswer":null,"CompanyId":"6edc1973f7a24c65a305a6db001f52d0","QuestionID":"8a818149743001ce01744401d95b4ad2","IsAnswer":1,"SortID":null,"Id":"64542a55-288e-44ce-92b3-d4a99de4e6c6"},{"AnswerChoice":".csv","SubjectiveAnswer":null,"CompanyId":"6edc1973f7a24c65a305a6db001f52d0","QuestionID":"8a818149743001ce01744401d95b4ad2","IsAnswer":1,"SortID":null,"Id":"df1d3c79-345c-4e07-9fef-724d4d0aeb38"}],"CreateTime":1598868806000,"TestTime":null,"QuestionBankId":"8a818149743001ce01744401d85e4ad1","QuestionType":5,"TotalCount":null,"BorderLine":null,"CreateUserId":"8a81c08b718291ba01720bc3ac182701","Answer":null,"UserAnswer":null,"Score":2,"EmployeeCoursePlanId":"1314462768265154560","CourseId":"1300376744474755072","AnswerChoice":null,"QuestionId":"8a818149743001ce01744401d95b4ad2","Content":"请给根据以下的药品信息，完成对乙酰氨基酚的WHOdrug编码。\n商品名：Dymadon；\n通用名称：Paracetamol；\n国家：Australia；\n剂型：Tablets；\n则该药品的drug code为 （）；该药品的MPID为（）。","Analysis":null,"ChoiceAnswer":"EE6L8MpgJqFLzvePaSz8Iu+gD2fNjG9QQUi5TNpjpQo=","UpdateTime":1598868806000,"UpdateUserId":"8a81c08b718291ba01720bc3ac182701","Id":"8a818149743001ce01744401d95b4ad2","SortAnswer":null,"IsTest":1,"KnowledgepointId":null,"IsCorrect":null}],"TotalCount":1}
\ No newline at end of file

pom.xml

@@ -8,9 +8,10 @@
 	<parent>
 		<groupId>net.wicp.tams</groupId>
 		<artifactId>common</artifactId>
-		<version>3.6.13</version>
+		<version>3.6.14</version>
 	</parent>
 	<modules>
 		<module>duckula3-dev-demo-cuslistener</module>
+		<module>demo-dev-kafka</module>
 	</modules>
 </project>
\ No newline at end of file