commit

deploy/binlog.sh

@@ -2,8 +2,10 @@ cd `dirname $0`
 cd ../
 
 version=$(date +%s)
-docker build -t registry.cn-chengdu.aliyuncs.com/cxzx-spider/taide-binlog:$version .
-docker push registry.cn-chengdu.aliyuncs.com/cxzx-spider/taide-binlog:$version
-sed -i "s!{ { image } }!registry\.cn-chengdu\.aliyuncs\.com/cxzx-spider/taide-binlog:$version!g" deploy/taide-binlog.yml
+docker build -t registry.cn-chengdu.aliyuncs.com/cxzx-spider/binlog:$version .
+docker push registry.cn-chengdu.aliyuncs.com/cxzx-spider/binlog:$version
+sed -i "s!{ { image } }!registry\.cn-chengdu\.aliyuncs\.com/cxzx-spider/binlog:$version!g" deploy/taide-binlog.yml
+sed -i "s!{ { image } }!registry\.cn-chengdu\.aliyuncs\.com/cxzx-spider/binlog:$version!g" deploy/huaqiyun-binlog.yml
 
 kubectl apply -f deploy/taide-binlog.yml
+kubectl apply -f deploy/huaqiyun-binlog.yml

deploy/huaqiyun-binlog.yml

@@ -0,0 +1,51 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: huaqiyun-binlog
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      k8s-app: huaqiyun-binlog
+  strategy:
+    type: Recreate
+  template:
+    metadata:
+      labels:
+        k8s-app: huaqiyun-binlog
+    spec:
+      hostAliases:
+        - ip: "172.16.101.16"
+          hostnames:
+            - "baidu06"
+        - ip: "172.16.101.17"
+          hostnames:
+            - "baidu07"
+        - ip: "172.16.101.18"
+          hostnames:
+            - "baidu08"
+        - ip: "172.16.101.19"
+          hostnames:
+            - "baidu09"
+      imagePullSecrets:
+        - name: registry-key
+      containers:
+        - command: [ "java","-cp","/app.jar","com.cxzx.canal.CanalClientMain","huaqiyun","huaqiyun-binlog" ]
+          env:
+            - name: PATH
+              value: /usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
+            - name: TZ
+              value: Asia/Shanghai
+          image: { { image } }
+          imagePullPolicy: IfNotPresent
+          name: huaqiyun-binlog
+          resources:
+            limits:
+              cpu: 100m
+              memory: 515Mi
+            requests:
+              cpu: 100m
+              memory: 50Mi
+      dnsPolicy: ClusterFirst
+      terminationGracePeriodSeconds: 10
+

deploy/taide-binlog.yml

@@ -30,7 +30,7 @@ spec:
       imagePullSecrets:
         - name: registry-key
       containers:
-        - command: [ "java","-cp","/app.jar","com.cxzx.canal.CanalClientMain" ]
+        - command: [ "java","-cp","/app.jar","com.cxzx.canal.CanalClientMain","canal","taide-binlog" ]
           env:
             - name: PATH
               value: /usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin

src/main/java/com/cxzx/canal/CanalClientMain.java

@@ -1,19 +1,17 @@
 package com.cxzx.canal;
 
-import java.util.List;
-import java.util.concurrent.TimeUnit;
-
-import com.alibaba.fastjson2.JSONArray;
 import com.alibaba.fastjson2.JSONObject;
+import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
 import com.alibaba.otter.canal.protocol.CanalEntry;
+import com.alibaba.otter.canal.protocol.Message;
 import com.aliyun.openservices.log.common.LogItem;
 import com.cxzx.sls.AliyunLogger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.util.Assert;
 
-import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector;
-import com.alibaba.otter.canal.protocol.Message;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 public class CanalClientMain {
     protected final static Logger logger = LoggerFactory.getLogger(CanalClientMain.class);
@@ -28,18 +26,22 @@ public class CanalClientMain {
 
     private final Thread.UncaughtExceptionHandler handler = (t, e) -> logger.error("parse events has an error", e);
 
-    public CanalClientMain(String servers, String topic, Integer partition, String groupId) {
+    public CanalClientMain(String servers, String topic, Integer partition, String groupId, String logStore) {
         connector = new KafkaCanalConnector(servers, topic, partition, groupId, null, false);
-        sls = new AliyunLogger("taide-binlog");
+        sls = new AliyunLogger(logStore);
     }
 
     public static void main(String[] args) {
+        // canal taide-binlog # huaqiyun huaqiyun-binlog
+        var topic = args[0];
+        var logStore = args[1];
         try {
             final CanalClientMain canalClient = new CanalClientMain(
                     "baidu06:9092,baidu07:9092,baidu08:9092,baidu09:9092",
-                    "canal",
+                    topic,
                     0,
-                    "zyx-canal");
+                    "zyx-canal",
+                    logStore);
             canalClient.start();
             logger.info("## the canal kafka consumer is running now ......");
             Runtime.getRuntime().addShutdownHook(new Thread(() -> {