完善配置文件

.idea/.gitignore

@@ -0,0 +1,8 @@
+# Default ignored files
+/shelf/
+/workspace.xml
+# Editor-based HTTP Client requests
+/httpRequests/
+# Datasource local storage ignored files
+/dataSources/
+/dataSources.local.xml

.idea/ent_dispose.iml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="WEB_MODULE" version="4">
+  <component name="Go" enabled="true" />
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/ent_dispose.iml" filepath="$PROJECT_DIR$/.idea/ent_dispose.iml" />
+    </modules>
+  </component>
+</project>

.idea/vcs.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>

README.md

@@ -0,0 +1,38 @@
+## 新版程序
+
+### 1.程序udp 端口占用
+
+| 程序          | 端口  | 备注                                               |
+| ------------- | ----- | -------------------------------------------------- |
+| qyxy_listen   | 18887 | 监听文件夹变化                                     |
+| qyxy_inc_data | 18888 | 对应原来inx_data程序，更新所有数据表               |
+| qyxy_inc_std  | 18889 | 对应原来qyxy_std，读取新增数据更新到qyxy_std数据表 |
+| qyxy_es       | 18890 | 读取qyxy_std 新增数据存入es                        |
+| qyxy_change   | 18891 | 更新企业变更数据                                   |
+
+
+
+### 2.程序介绍
+
+1. `	qyxy_listen`： 负责监听上传目录，获取最新的文件夹，然后传递路径给`inc_data`、`inc_std`程序。
+
+    
+
+2. `qyxy_inc_data`: 读取目录下的所有文件夹，根据表名增量同步数据到各个表,操作的数据库为**172.17.4.181**； 程序执行完毕通知`qyxy_change `程序，更新`company_change`数据内容到 `qyxy_change`中，配置文件大致如下：
+
+   | local_port       | udp 本地监听端口，本地监听端口需要保留`:` |      |
+   | ---------------- | ----------------------------------------- | ---- |
+   | qyxy_change_port | qyxy_change_port 监听的udp 端口           |      |
+   | target_ip        | udp 目标地址IP，默认本地                  |      |
+   | mgoAddr          | MongoDB 地址                              |      |
+   | db_name          | MongoDB 数据库                            |      |
+
+3. `qyxy_std`: 读取目录下所有文件，主要读取6个数据表的内容，然后更新到**172.17.145.163 **的`mixdata `数据库`qyxy_std`数据表.
+
+4. `qyxy_es`:  负责读取**172.17.145.163**的 `qyxy_std`数据表迭代数据，更新到 **172.17.4.184:19800** es 索引中。
+
+5. `qyxy_change`：主要同步企业增量变更数据，从`181`的`company_change`表获取数据，然后更新到`172.17.145.163:27083`的数据表`qyxy_change`。获取数据大于配置文件`_id_`数据，如果配置文件配置了`lastId`,就使用配置文件的，否则使用udp 信号传递过来的`last_id`。
+
+
+
+ 

Readme.md

@@ -1,19 +0,0 @@
-## 新版程序
-
-### 1.程序udp 端口占用
-
-| 程序          | 端口  | 备注                                               |
-| ------------- | ----- | -------------------------------------------------- |
-| qyxy_listen   | 18887 | 监听文件夹变化                                     |
-| qyxy_inc_data | 18888 | 对应原来inx_data程序，更新所有数据表               |
-| qyxy_inc_std  | 18889 | 对应原来qyxy_std，读取新增数据更新到qyxy_std数据表 |
-| qyxy_es       | 18890 | 读取qyxy_std 新增数据存入es                        |
-| qyxy_change   | 18891 | 更新企业变更数据                                   |
-
-
-
-### 2.程序介绍
-
-1. `qyxy_change`：主要同步企业变更数据，从`181`的`company_change`表获取数据，获取数据大于配置文件`_id_`数据，如果配置文件配置了`lastId`,就使用配置文件的，否则使用udp 信号传递过来的`last_id`。
-2. ``
-

qyxy_change_new/config.json

@@ -1,14 +1,14 @@
 {
-  "dbServer": "192.168.3.207:27092",
-  "dbName": "wcc",
-  "dbSize": 3,
-  "uname": "",
-  "upwd": "",
+  "dbServer": "172.17.4.187:27082,172.17.145.163:27083",
+  "dbName": "mixdata",
+  "dbSize": 10,
+  "uname": "SJZY_RWMIX_Other",
+  "upwd": "SJZY@M34I6x7D9ata",
   "coll_qy": "qyxy_std",
   "coll_change": "qyxy_change",
   "local_port": ":18891",
   "lastId": 0,
-  "company_server": "127.0.0.1:27001",
+  "company_server": "172.17.4.181:27001",
   "company_db": "mixdata",
   "changeType": [
     {

qyxy_change_new/main.go

@@ -70,8 +70,8 @@ func initChangeMap() {
 }
 
 func main() {
-	//UdpClient.Listen(processUdpMsg)
-	//util.Debug("Udp服务监听======= port:", localPort)
+	UdpClient.Listen(processUdpMsg)
+	util.Debug("Udp服务监听======= port:", localPort)
 
 	ch := make(chan bool, 1)
 	<-ch

qyxy_change_new/test.json

@@ -0,0 +1,92 @@
+{
+  "dbServer": "192.168.3.207:27092",
+  "dbName": "wcc",
+  "dbSize": 3,
+  "uname": "",
+  "upwd": "",
+  "coll_qy": "qyxy_std",
+  "coll_change": "qyxy_change",
+  "local_port": ":18891",
+  "lastId": 0,
+  "company_server": "192.168.3.207:27092",
+  "company_db": "mixdata",
+  "changeType": [
+    {
+      "change_code": "100001",
+      "change_name": "法定代表人变更",
+      "change_push": true,
+      "change_info": "法定代表人变更",
+      "change_keyword": ["法定代表人", "法人"]
+    },
+    {
+      "change_code": "100002",
+      "change_name": "注册资本变更",
+      "change_push": true,
+      "change_info": "注册资本",
+      "change_keyword": ["注册资本"]
+    },
+    {
+      "change_code": "100003",
+      "change_name": "企业名称变更",
+      "change_push": true,
+      "change_info": "注册资本变更（注册资金、资金数额等变更）变更",
+      "change_keyword": ["企业名称", "隶属企业", "集团名称"]
+    },
+    {
+      "change_code": "100004",
+      "change_name": "投资人变更",
+      "change_push": true,
+      "change_info": "投资人、合伙人、投资者等名称变更",
+      "change_keyword": ["投资人", "合伙人", "投资者", "董事", "经营者", "股东"]
+    },
+    {
+      "change_code": "100005",
+      "change_name": "经营范围变更",
+      "change_push": true,
+      "change_info": "经营范围变更",
+      "change_keyword": ["经营范围", "业务范围", "许可经营项目", "隶属企业变更", "母公司变更", "隶属集团变更"]
+    },
+    {
+      "change_code": "100006",
+      "change_name": "经营期限变更",
+      "change_push": false,
+      "change_info": "经营期限",
+      "change_keyword": ["经营期限", "营业期限", "驻在期限", "合伙期限"]
+    },
+    {
+      "change_code": "100007",
+      "change_name": "高级管理人员备案",
+      "change_push": true,
+      "change_info": "投资人、董事、高管",
+      "change_keyword": ["高级管理人员", "高管人员", "经理", "主要人员", "负责人", "代表", "理事", "监事", "委员", "主要部门人员"]
+    },
+    {
+      "change_code": "100008",
+      "change_name": "其他事项备案",
+      "change_push": false,
+      "change_info": "其他事项备案",
+      "change_keyword": ["备案"]
+    },
+    {
+      "change_code": "100009",
+      "change_name": "章程备案变更",
+      "change_push": false,
+      "change_info": "章程备案变更",
+      "change_keyword": ["章程修正", "章程备案", "合伙协议"]
+    },
+    {
+      "change_code": "100010",
+      "change_name": "经营场所，地址变更",
+      "change_push": false,
+      "change_info": "经营场所，地址变更",
+      "change_keyword": ["营业地址", "地址", "住所", "经营产所"]
+    },
+    {
+      "change_code": "100011",
+      "change_name": "其他变更",
+      "change_push": false,
+      "change_info": "无法分类的变更",
+      "change_keyword": []
+    }
+  ]
+}

qyxy_es_new/config.json

@@ -1,7 +1,7 @@
 {
-  "mgodb": "127.0.0.1:27017",
+  "mgodb": "172.17.4.187:27082,172.17.145.163:27083",
   "dbsize": 12,
-  "dbname": "wcc",
+  "dbname": "mixdata",
   "dbcoll": "qyxy_std",
   "uname": "",
   "upwd": "",
@@ -9,8 +9,8 @@
   "updatetime": 0,
   "local_port": ":18890",
   "elastic": {
-    "addr": "http://192.168.3.206:9800",
-    "index": "qyxy",
+    "addr": "http://172.17.4.184:19800",
+    "index": "qyxy_v1",
     "itype": "qyxy",
     "pool": 12,
     "esfields": [

qyxy_es_new/test.json

@@ -0,0 +1,62 @@
+{
+  "mgodb": "192.168.3.207:27092",
+  "dbsize": 12,
+  "dbname": "wcc",
+  "dbcoll": "qyxy_std",
+  "uname": "",
+  "upwd": "",
+  "tasktime": 0,
+  "updatetime": 0,
+  "local_port": ":18890",
+  "elastic": {
+    "addr": "http://192.168.3.206:9800",
+    "index": "qyxy_v1",
+    "itype": "qyxy",
+    "pool": 12,
+    "esfields": [
+      "_id",
+      "company_name",
+      "history_name",
+      "credit_no",
+      "org_code",
+      "tax_code",
+      "company_code",
+      "company_type",
+      "company_type_old",
+      "company_area",
+      "company_city",
+      "company_district",
+      "company_phone",
+      "company_email",
+      "updatetime",
+      "legal_person",
+      "establish_date",
+      "lastupdatetime",
+      "capital",
+      "currency",
+      "operation_startdate",
+      "operation_enddate",
+      "authority",
+      "issue_date",
+      "company_status",
+      "company_address",
+      "business_scope",
+      "cancel_date",
+      "revoke_date",
+      "employee_name",
+      "stock_name",
+      "bid_projectname",
+      "bid_purchasing",
+      "bid_area",
+      "bid_unittype",
+      "bid_contracttype",
+      "search_type",
+      "company_shortname",
+      "website_url",
+      "partners",
+      "employees",
+      "tag_business",
+      "employee_num"
+    ]
+  }
+}

qyxy_listen/config.json

@@ -5,5 +5,6 @@
   "qyxy_es_port": "18890",
   "qyxy_change_port": "18891",
   "target_ip": "127.0.0.1",
-  "watch_path": "/Users/wangchengcheng/Desktop/jianyu/upload/"
+  "watch_path": "/Users/wangchengcheng/Desktop/jianyu/upload/",
+  "manual_path": ""
 }

qyxy_listen/go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/andybalholm/cascadia v1.3.1 // indirect
 	github.com/dchest/captcha v1.0.0 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
 	golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2 // indirect
 	golang.org/x/sys v0.0.0-20220908164124-27713097b956 // indirect
 	gopkg.in/mgo.v2 v2.0.0-20190816093944-a6b53ec6cb22 // indirect

qyxy_listen/go.sum

@@ -26,6 +26,8 @@ github.com/nsqio/go-nsq v1.1.0/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQT
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=

qyxy_listen/main.go

@@ -5,7 +5,7 @@ import (
 	"app.yhyue.com/data_processing/common_utils/udp"
 	"encoding/json"
 	"fmt"
-	"github.com/fsnotify/fsnotify"
+	"github.com/robfig/cron/v3"
 	"log"
 	"net"
 	"os"
@@ -29,8 +29,6 @@ var (
 	LastPath   string //上传的最新文件夹
 )
 
-//var option utils.Option
-
 func init() {
 	util.ReadConfig(&Sysconfig)
 	localPort := Sysconfig["local_port"].(string)
@@ -62,173 +60,104 @@ func init() {
 }
 
 func main() {
-	//监听目录
-	watchPath := util.ObjToString(Sysconfig["watch_path"])
-	pathinfo, _ := os.Stat(watchPath)
-	if pathinfo.IsDir() {
-		//先读取upload 所有文件夹
-		dir := utils.Option{
-			RootPath:   watchPath,
-			SubFlag:    false,
-			IgnorePath: []string{`.git`, `.svn`},
-			IgnoreFile: []string{`.DS_Store`, `.gitignore`},
-		}
-		dirNodes, err := utils.Explorer(dir)
-		if err != nil {
-			util.Debug("utils.Explorer err =>", err)
-		}
+	local, _ := time.LoadLocation("Asia/Shanghai")
+	cronServer := cron.New(cron.WithLocation(local), cron.WithSeconds())
 
-		//获取三天内新建的文件夹，表示需要读取 的文件夹
-		for _, v := range dirNodes.Children {
-			if v.ModTime.AddDate(0, 0, 3).Sub(time.Now()) > 0 {
-				LastPath = v.Path
-			}
-		}
+	//spec := "0 */1 * * * *" //秒，分，时，日，月，星期
+	spec := "0 57 09 * * 1"
+	cronServer.AddFunc(spec, dealWatch)
+	//c.AddFunc("0 9 * * 1", dealWatch)
 
-		option.RootPath = LastPath
-		rootNodes, err := utils.Explorer(option)
-		if err != nil {
-			util.Debug("utils.Explorer err =>", err)
-		}
+	cronServer.Start()
 
-		//1.判断数据表数量
-		if len(rootNodes.Children) != pathTableSize {
-			log.Printf("%s 目录下文件夹数量为：%s", rootNodes.Path, len(rootNodes.Children))
-		} else {
-			for _, v := range rootNodes.Children {
-				fmt.Println(v.Name)
-				if len(v.Children) != tableDateSize {
-					log.Printf("%s 目录下文件夹数量为：%s", v.Path, len(v.Children))
-				} else {
-					for _, vv := range v.Children {
-						for _, file := range vv.Children {
-							if file.Name == "split.json.gz" {
-								dateSlice = append(dateSlice, true)
-							}
-						}
-					}
-				}
-				if len(dateSlice) == tableDateSize {
-					tableSlice = append(tableSlice, true)
-				}
-				dateSlice = make([]bool, 0)
+	select {}
+}
+
+func test() {
+	fmt.Println("test -----")
+	log.Println("start dealWatch")
+	fmt.Println("test >>>>>>")
+}
+
+//dealWatch 处理监听程序，获取最新的文件夹路径
+func dealWatch() {
+	log.Println("start dealWatch")
+	//手动配置的文件路径
+	manualPath := util.ObjToString(Sysconfig["manual_path"])
+	if manualPath == "" {
+		//监听目录
+		watchPath := util.ObjToString(Sysconfig["watch_path"])
+		pathinfo, _ := os.Stat(watchPath)
+		if pathinfo.IsDir() {
+			//先读取upload 所有文件夹
+			dir := utils.Option{
+				RootPath:   watchPath,
+				SubFlag:    false,
+				IgnorePath: []string{`.git`, `.svn`},
+				IgnoreFile: []string{`.DS_Store`, `.gitignore`},
+			}
+			dirNodes, err := utils.Explorer(dir)
+			if err != nil {
+				util.Debug("utils.Explorer err =>", err)
 			}
 
-			//满足条件，触发同步
-			if len(tableSlice) == pathTableSize {
-				data := map[string]interface{}{
-					"start": true,
-					"path":  LastPath,
+			//获取三天内新建的文件夹，表示需要读取 的文件夹
+			for _, v := range dirNodes.Children {
+				if v.ModTime.AddDate(0, 0, 3).Sub(time.Now()) > 0 {
+					LastPath = v.Path
 				}
-				SendUdpMsg(data, incDataAddr) //inc_data
-				SendUdpMsg(data, incStdAddr)  // inc_std
 			}
-			tableSlice = make([]bool, 0)
+		} else {
+			LastPath = manualPath
 		}
 
+		//真正读取压缩包
+		dealFiles()
 	}
 }
 
-func main2() {
-	// Create new watcher.
-	watcher, err := fsnotify.NewWatcher()
+//dealFiles 处理路径下的文件
+func dealFiles() {
+	option.RootPath = LastPath
+	rootNodes, err := utils.Explorer(option)
 	if err != nil {
-		log.Fatal("NewWatcher :", err)
+		util.Debug("utils.Explorer err =>", err)
 	}
-	defer watcher.Close()
 
-	// 添加监听目录
-	err = watcher.Add(Sysconfig["watch_path"].(string))
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	// Start listening for events.
-	go func() {
-		for {
-			select {
-			case event, ok := <-watcher.Events:
-				if !ok {
-					return
-				}
-				//判断事件发生的类型，如下5种
-				// Create 创建
-				// Write 写入
-				// Remove 删除
-				// Rename 重命名
-				// Chmod 修改权限
-				if event.Op&fsnotify.Create == fsnotify.Create {
-					log.Println("创建文件 : ", event.Name)
-					option.RootPath = event.Name
-					pathinfo, _ := os.Stat(event.Name)
-					if pathinfo.IsDir() {
-						rootNodes, err := utils.Explorer(option)
-						if err != nil {
-							util.Debug("utils.Explorer err =>", err)
+	//1.判断数据表数量
+	if len(rootNodes.Children) != pathTableSize {
+		log.Printf("%s 目录下文件夹数量为：%s", rootNodes.Path, len(rootNodes.Children))
+	} else {
+		for _, v := range rootNodes.Children {
+			fmt.Println(v.Name)
+			if len(v.Children) != tableDateSize {
+				log.Printf("%s 目录下文件夹数量为：%s", v.Path, len(v.Children))
+			} else {
+				for _, vv := range v.Children {
+					for _, file := range vv.Children {
+						if file.Name == "split.json.gz" {
+							dateSlice = append(dateSlice, true)
 						}
-						//1.判断数据表数量
-						if len(rootNodes.Children) != pathTableSize {
-							log.Printf("%s 目录下文件夹数量为：%s", rootNodes.Path, len(rootNodes.Children))
-						} else {
-							for _, v := range rootNodes.Children {
-								fmt.Println(v.Name)
-								if len(v.Children) != tableDateSize {
-									log.Printf("%s 目录下文件夹数量为：%s", v.Path, len(v.Children))
-								} else {
-									for _, vv := range v.Children {
-										for _, file := range vv.Children {
-											if file.Name == "split.json.gz" {
-												dateSlice = append(dateSlice, true)
-											}
-										}
-									}
-								}
-								if len(dateSlice) == tableDateSize {
-									tableSlice = append(tableSlice, true)
-								}
-								dateSlice = make([]bool, 0)
-							}
-
-							//fmt.Println("aaaa=>", len(tableSlice))
-							//满足条件，触发同步
-							if len(tableSlice) == pathTableSize {
-								data := map[string]interface{}{
-									"start": true,
-									"path":  event.Name,
-								}
-								SendUdpMsg(data, incDataAddr)
-								SendUdpMsg(data, incStdAddr)
-							}
-							tableSlice = make([]bool, 0)
-						}
-
 					}
 				}
-
-				if event.Op&fsnotify.Write == fsnotify.Write {
-					log.Println("写入文件 : ", event.Name)
-				}
-				if event.Op&fsnotify.Remove == fsnotify.Remove {
-					log.Println("删除文件 : ", event.Name)
-				}
-				if event.Op&fsnotify.Rename == fsnotify.Rename {
-					log.Println("重命名文件 : ", event.Name)
-				}
-				if event.Op&fsnotify.Chmod == fsnotify.Chmod {
-					log.Println("修改权限 : ", event.Name)
-				}
-			case err, ok := <-watcher.Errors:
-				if !ok {
-					return
-				}
-				log.Println("error:", err)
 			}
+			if len(dateSlice) == tableDateSize {
+				tableSlice = append(tableSlice, true)
+			}
+			dateSlice = make([]bool, 0)
 		}
-	}()
-
-	// Block main goroutine forever.
-	<-make(chan struct{})
 
+		//满足条件，触发同步
+		if len(tableSlice) == pathTableSize {
+			data := map[string]interface{}{
+				"start": true,
+				"path":  LastPath,
+			}
+			SendUdpMsg(data, incDataAddr) //inc_data
+			SendUdpMsg(data, incStdAddr)  // inc_std
+		}
+		tableSlice = make([]bool, 0)
+	}
 }
 
 //SendUdpMsg 通知处理企业新增数据
@@ -239,6 +168,7 @@ func SendUdpMsg(data map[string]interface{}, target *net.UDPAddr) {
 	util.Debug("target :=>", target.IP, target.Port)
 }
 
+//processUdpMsg 处理udp
 func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 	switch act {
 	case udp.OP_TYPE_DATA:

qyxy_std_new/config.json

@@ -1,10 +1,10 @@
 {
-  "dbServer": "127.0.0.1:27017",
-  "dbName": "wcc",
+  "dbServer": "172.17.4.187:27082,172.17.145.163:27083",
+  "dbName": "mixdata",
   "dbSize": 10,
   "dbSave": "qyxy_std",
-  "uname": "",
-  "upwd": "",
+  "uname": "SJZY_RWMIX_Other",
+  "upwd": "SJZY@M34I6x7D9ata",
   "lastId": 0,
   "local_port": ":18889",
   "qyxy_es_port": "18890",

qyxy_std_new/main.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"log"
 	"net"
 	"os"
 	"strings"
@@ -351,7 +352,7 @@ func dealCompanyBase(data map[string]interface{}) {
 		tmps := saveArr
 		res := MongoTool.UpSertBulk(Sysconfig["dbSave"].(string), tmps...)
 		if !res {
-			fmt.Println("eeeeeeeeeeeeeeeeeeeeeeeeeeeee")
+			log.Println("UpSertBulk  company_base err")
 		}
 		saveArr = [][]map[string]interface{}{}
 	}

qyxy_std_new/test.json

@@ -0,0 +1,13 @@
+{
+  "dbServer": "192.168.3.207:27092",
+  "dbName": "wcc",
+  "dbSize": 10,
+  "dbSave": "qyxy_std",
+  "uname": "",
+  "upwd": "",
+  "lastId": 0,
+  "local_port": ":18889",
+  "qyxy_es_port": "18890",
+  "target_ip": "127.0.0.1",
+  "path": ""
+}

qyy_inc_data_new/config.json

@@ -3,6 +3,6 @@
   "qyxy_change_port": "18891",
   "target_ip": "127.0.0.1",
   "path": "",
-  "mgoAddr": "127.0.0.1:27017",
+  "mgoAddr": "172.17.4.181:27001",
   "db_name": "mixdata"
 }

qyy_inc_data_new/main.go

@@ -8,8 +8,10 @@ import (
 	"compress/gzip"
 	"encoding/json"
 	"fmt"
+	"go.uber.org/zap"
 	"io"
 	"io/ioutil"
+	"log"
 	"net"
 	"os"
 	"strings"
@@ -36,7 +38,7 @@ var (
 	changePort string // qyxy_change 监听端口
 	changeAddr *net.UDPAddr
 	readPath   string //文件夹目录
-	lastID     int64
+	//lastID     int64
 )
 
 func init() {
@@ -70,8 +72,8 @@ func init() {
 func main() {
 	UdpClient = udp.UdpClient{Local: localPort, BufSize: 1024}
 	UdpClient.Listen(processUdpMsg)
-	//util.Debug(localPort)
-	//fmt.Println("Udp服务监听=======", zap.String("port:", localPort))
+
+	log.Println("Udp服务监听=======", zap.String("port:", localPort))
 	util.Debug("Udp服务监听======= port: ", localPort)
 
 	ch := make(chan bool, 1)
@@ -88,7 +90,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 		//拿到同步信号，开始同步数据
 		if _, ok := mapInfo["start"]; ok {
 			if _, okk := mapInfo["path"]; okk {
-				path := util.ObjToString(mapInfo["path"])
+				path := util.ObjToString(mapInfo["path"]) //udp 传递的路径
 				//没有指定配置文件的指定目录，就使用udp 传递目录
 				if readPath == "" && path != "" {
 					readPath = path

qyy_inc_data_new/test.json

@@ -0,0 +1,8 @@
+{
+  "local_port": ":18888",
+  "qyxy_change_port": "18891",
+  "target_ip": "127.0.0.1",
+  "path": "",
+  "mgoAddr": "192.168.3.207:27092",
+  "db_name": "mixdata"
+}