update: add batch size

Author: li-phone

fake-stream/fakestream/FakeStreamWorker.py

@@ -12,18 +12,18 @@ def dump_log(line, log_path=f'logs/{fileftime()}.log'):
 
 
 class FakeStreamWorker(AbstractParallel, ABC):
-    def __init__(self, log_func, regex_params, **kwargs):
+    def __init__(self, log_func, regex_params, batch_size, **kwargs):
         super().__init__(**kwargs)
         self.print_log = log_func
         self.regex_params = regex_params
+        self.batch_size = batch_size
         self.host = None
         self.dbname = None
-        self.workers_num = 1
-        self.connection = None
         self.fix_val = None
-        self.time_offset = None
         self.speed = None
+        self.workers_num = 1
         self.table_names = None
+        self.connection = None
         self.run_flag = None
         self.table_map = None
 
@@ -57,16 +57,19 @@ def process_task(self, task, **kwargs):
                             self.table_map[table_name] = update_table(
                                 exec_sql(f"desc {table_name}", self.connection, reshape=False), self.regex_params)
                 for k, v in self.table_map.items():
-                    fs = fill_table(v, self.regex_params, self.host, self.dbname, k, default_regex=self.fix_val)
-                    ret, res = add([fs], k, self.connection, freq=1)
+                    records = []
+                    for b in range(self.batch_size):
+                        record = fill_table(v, self.regex_params, self.host, self.dbname, k, default_regex=self.fix_val)
+                        records.append(record)
+                    ret, res = add(records, k, self.connection, freq=len(records))
                     if ret:
-                        success_cnt += 1
+                        success_cnt += len(records)
                     else:
-                        failure_cnt += 1
+                        failure_cnt += len(records)
                     for line in res:
                         self.print_log(LOGGER.info(f'execute sql "{line}"'))
             except Exception as e:
-                failure_cnt += 1
+                failure_cnt += self.batch_size
                 self.print_log(LOGGER.error(e.args))
                 self.print_log(LOGGER.error(f'execute failure!'))
 

fake-stream/fakestream/MainApp.py

@@ -45,7 +45,7 @@ def __init__(self, parent, title, size, *args, **kwargs):
 
         self.regex_params = read_bin('regex-parameters.json')
         self.parallel = FakeStreamWorker(is_join=False, print_process=False, log_func=self.print_log,
-                                         regex_params=self.regex_params)
+                                         regex_params=self.regex_params, batch_size=1)
         self.db_params = read_bin(self.db_parameter_path)
         self.db_params = {} if self.db_params is None else self.db_params
         self.sorted_params = []
@@ -183,9 +183,9 @@ def InitUI(self):
         paramGrid = wx.GridBagSizer(0, 0)
         self.fixValST = wx.StaticText(panel, -1, "设定值")
         self.fixValTC = wx.TextCtrl(panel, -1, style=wx.ALIGN_LEFT, value='6')
-        self.offsetST = wx.StaticText(panel, -1, "时间补偿(秒)")
-        self.offsetTC = wx.TextCtrl(panel, -1, style=wx.ALIGN_LEFT, value='0',
-                                    validator=NumberValidator(DataType.FLOAT))
+        self.offsetST = wx.StaticText(panel, -1, "批大小")
+        self.offsetTC = wx.TextCtrl(panel, -1, style=wx.ALIGN_LEFT, value='1',
+                                    validator=NumberValidator(DataType.UINT))
         self.workerST = wx.StaticText(panel, -1, "线程数")
         self.workerTC = wx.TextCtrl(panel, -1, style=wx.ALIGN_LEFT, value='1',
                                     validator=NumberValidator(DataType.UINT))
@@ -265,8 +265,8 @@ def OnClickConnect(self, event):
     def OnClickRun(self, event):
         self.OnClickConnect(event)
         run_params = dict(
+            batch_size=int(self.offsetTC.GetValue()),
             fix_val=self.fixValTC.GetValue(),
-            time_offset=float(self.offsetTC.GetValue()),
             speed=float(self.speedsTC.GetValue()),
             workers_num=int(self.workerTC.GetValue()),
             table_names=self.tbNameTC.GetValue().split(),

fake-stream/fakestream/regex-parameters.json

@@ -1,8 +1,8 @@
 {
-  "Name.*.*.*.*": "Faker::fake.name()",
-  "CountryCode.*.*.*.*": "Faker::fake.country_code()",
-  "District.*.*.*.*": "Faker::fake.district()",
+  "name.*.*.*.*": "::fake.name()",
+  "countryCode.*.*.*.*": "::fake.country_code()",
+  "district.*.*.*.*": "::fake.district()",
   "*.*.*.*.int": "6+",
-  "citycol.*.*.*.*": "Faker::fake.word()",
+  "citycol.*.*.*.*": "::fake.word()",
   "*.*.*.*.datetime": "::time.strftime('%Y-%m-%d %H:%M:%S.%m')"
 }