कैसे जाने के साथ mysql binlog पकाने के लिए

मेरा नाम एर्टोम है, मैं गो लीड डेवलपर की स्थिति में प्रोजेक्ट "स्ट्रीम" में रैम्बलर ग्रुप में काम करता हूं।
हमने बहुत समय बिताया, मैक्कल बिनलॉग को तैयार करने में। यह आलेख एक कहानी है कि कैसे जल्दी से और न्यूनतम संख्या में नुकसान के साथ गो बिनलॉग के साथ काम करने के लिए तंत्र को लागू करें।

हमें इसकी आवश्यकता क्यों है?

स्ट्रीम के हुड के तहत अत्यधिक लोड किए गए मॉड्यूल हैं, जहां डेटाबेस में प्रत्येक क्वेरी उपयोगकर्ता को परिणाम प्राप्त करने से दूर ले जाती है। कैशिंग एक अच्छा समाधान है, लेकिन कैश को कब फ्लश करना है? डेटा खुद बताएं कि वे अपडेट किए गए हैं।

Mysql में मास्टर-स्लेव प्रतिकृति के रूप में ऐसी चीज है। हमारा डेमन एक दास होने का दिखावा कर सकता है और बिनलोग के माध्यम से डेटा प्राप्त कर सकता है। बिनलॉग को पंक्ति प्रारूप में कॉन्फ़िगर किया जाना चाहिए। इसमें सभी डेटाबेस परिवर्तन कमांड शामिल हैं, लेन-देन के तहत आदेश प्रतिबद्ध होने के बाद ही निष्पादित किए जाते हैं। अधिकतम अनुमत आकार (डिफ़ॉल्ट रूप से 1 गीगा) तक पहुंचने पर, निम्न फ़ाइल बनाई जाती है। प्रत्येक नई फ़ाइल में नाम के बाद एक सीरियल नंबर होता है।

यहाँ या यहाँ थोड़ी अधिक जानकारी।

लेख के दो भाग हैं:

1. लॉग में प्राप्त प्रविष्टियों को कैसे जल्दी से संसाधित करना शुरू करें।
2. हुड के तहत क्या अनुकूलित और विस्तारित करना है।

भाग 1. हम जल्द से जल्द शुरू करते हैं।

बिनलॉग के साथ काम करने के लिए, हम github.com/siddontang/go-mysql लाइब्रेरी का उपयोग करेंगे
नए चैनल से कनेक्ट करें (चैनलों के साथ काम करने के लिए, बिनलॉग के लिए आरओडब्ल्यू प्रारूप की आवश्यकता है )।

func binLogListener() { c, err := getDefaultCanal() if err == nil { coords, err := c.GetMasterPos() if err == nil { c.SetEventHandler(&binlogHandler{}) c.RunFrom(coords) } } } func getDefaultCanal() (*canal.Canal, error) { cfg := canal.NewDefaultConfig() cfg.Addr = fmt.Sprintf("%s:%d", "127.0.0.1", 3306) cfg.User = "root" cfg.Password = "root" cfg.Flavor = "mysql" cfg.Dump.ExecutionPath = "" return canal.NewCanal(cfg) } 

बिनलोग के ऊपर एक आवरण बनाएँ:

 type binlogHandler struct { canal.DummyEventHandler //     BinlogParser //      } func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {return nil} func (h *binlogHandler) String() string {return "binlogHandler"} 

BinlogParser

हम OnRow () विधि में तर्क जोड़कर परिणामी Binlog लाइन के साथ काम करने के तर्क का विस्तार करेंगे।

 func (h *binlogHandler) OnRow(e *canal.RowsEvent) error { var n int //  var k int //  switch e.Action { case canal.DeleteAction: return nil //    case canal.UpdateAction: n = 1 k = 2 case canal.InsertAction: n = 0 k = 1 } for i := n; i < len(e.Rows); i += k { key := e.Table.Schema + "." + e.Table.Name switch key { case User{}.SchemaName() + "." + User{}.TableName(): /*    */ } } return nil } 

इस आवरण का सार प्राप्त आंकड़ों को पार्स करना है। लाइन को अपडेट करने के लिए डेटा दो रिकॉर्ड में आता है (पहली पंक्ति में मूल डेटा होगा, दूसरा - अपडेट किया गया)। यहां हम बहु-आवेषण और बहु-अद्यतन की संभावना पर भी विचार करते हैं। इस स्थिति में, हमें UPDATE के लिए हर दूसरा रिकॉर्ड लेना होगा। इसके लिए, ऊपर के उदाहरण में, हमने n और k में प्रवेश किया।

आइए, Binlog से डेटा प्राप्त करने के लिए एक मॉडल बनाएं। इसमें हम प्राप्त पंक्तियों से डेटा पढ़ेंगे। एनोटेशन में हम कॉलम के नाम दर्शाते हैं:

 type User struct { Id int `gorm:"column:id"` Name string `gorm:"column:name"` Status string `gorm:"column:status"` Created time.Time `gorm:"column:created"` } func (User) TableName() string { return "User" } func (User) SchemaName() string { return "Test" } 

MYSQL में तालिका संरचना:

 CREATE TABLE Test.User ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(40) NULL , status ENUM("active","deleted") DEFAULT "active", created TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP ) ENGINE =InnoDB; 

चलो स्वयं पार्सिंग का विश्लेषण करें - डेटा पार्सिंग के लिए तैयारी की जगह में जोड़ें:

 user := User{} h.GetBinLogData(&user, e, i) 

संक्षेप में, यह पर्याप्त है - हमारे पास उपयोगकर्ता मॉडल में नए रिकॉर्ड का डेटा होगा, लेकिन स्पष्टता के लिए, हम उन्हें प्रदर्शित करेंगे:

  if e.Action == canal.UpdateAction { oldUser := User{} h.GetBinLogData(&oldUser, e, i-1) fmt.Printf("User %d is updated from name %s to name %s\n", user.Id, oldUser.Name, user.Name, ) } else { fmt.Printf("User %d is created with name %s\n", user.Id, user.Name, ) } 

मुख्य बिंदु जिसके लिए हम प्रयास कर रहे थे, वह है हमारी "हैलो बिनलॉग वर्ल्ड" को लॉन्च करना:

 func main() { go binLogListener() //      time.Sleep(2 * time.Minute) fmt.Print("Thx for watching, goodbuy") } 

अगला, मान जोड़ें और अपडेट करें:

 INSERT INTO Test.User (`id`,`name`) VALUE (1,"Jack"); UPDATE Test.User SET name="Jonh" WHERE id=1; 

हम देखेंगे:

 User 1 is created with name Jack User 1 name changed from Jack to Jonh 

परिणामी कोड बिनलॉग और पार्स नई लाइनों के साथ काम करता है। जब हमें जिस तालिका की आवश्यकता होती है, उससे रिकॉर्ड प्राप्त करते हैं, तो कोड डेटा को संरचना में पढ़ता है और परिणाम प्रदर्शित करता है। पर्दे के पीछे डेटा पार्सर (BinlogParser) था, जिसने मॉडल को आबाद किया।

भाग 2. जैसा कि कॉब ने कहा, हमें एक स्तर कम चाहिए

पार्सर के आंतरिक काम पर विचार करें, जो प्रतिबिंब पर आधारित है।

मॉडल को डेटा से भरने के लिए, हमने हैंडलर विधि का उपयोग किया:

 h.GetBinLogData(&user, e, i) 

यह सरल डेटा प्रकारों को पार्स करता है:

 bool int float64 string time.Time 

और json से जटिल संरचनाओं को पार्स कर सकते हैं।

यदि समर्थित प्रकार आपके लिए पर्याप्त नहीं हैं, या आप बस यह समझना चाहते हैं कि बिनलॉग पार्सिंग कैसे काम करता है, तो आप अपने स्वयं के प्रकारों को जोड़ने का अभ्यास कर सकते हैं।

पहले, विचार करें कि आदर्श फ़ील्ड के Id फ़ील्ड के उदाहरण का उपयोग करके मॉडल फ़ील्ड के लिए डेटा कैसे भरें:

 type User struct { Id int `gorm:"column:id"` } 

प्रतिबिंब के माध्यम से हमें टाइप नाम मिलता है। ParseTagSetting विधि एनोटेशन को अधिक सुविधाजनक संरचना में परिवर्तित करती है:

 element := User{} //     ,      v := reflect.ValueOf(element) s := reflect.Indirect(v) t := s.Type() num := t.NumField() parsedTag := parseTagSetting(t.Field(k).Tag) if columnName, ok = parsedTag["COLUMN"]; !ok || columnName == "COLUMN" { continue } for k := 0; k < num; k++ { name := s.Field(k).Type().Name() switch name { case "int": //     } } 

अंतर प्रकार प्राप्त करने के बाद, आप प्रतिबिंब विधि के माध्यम से इसका मान निर्धारित कर सकते हैं:

 func (v Value) SetInt(x int64) {//... 

एनोटेशन को पार्स करने की विधि:

 func parseTagSetting(tags reflect.StructTag) map[string]string { setting := map[string]string{} for _, str := range []string{tags.Get("sql"), tags.Get("gorm")} { tags := strings.Split(str, ";") for _, value := range tags { v := strings.Split(value, ":") k := strings.TrimSpace(strings.ToUpper(v[0])) if len(v) >= 2 { setting[k] = strings.Join(v[1:], ":") } else { setting[k] = k } } } return setting } 

यह इनपुट के रूप में int64 को स्वीकार करता है। आइए एक ऐसी विधि बनाएं जो प्राप्त डेटा को बिनलॉग से इंट64 में अनुवाद करता है:

 func (m *BinlogParser) intHelper(e *canal.RowsEvent, n int, columnName string) int64 { columnId := m.getBinlogIdByName(e, columnName) if e.Table.Columns[columnId].Type != schema.TYPE_NUMBER { return 0 } switch e.Rows[n][columnId].(type) { case int8: return int64(e.Rows[n][columnId].(int8)) case int32: return int64(e.Rows[n][columnId].(int32)) case int64: return e.Rows[n][columnId].(int64) case int: return int64(e.Rows[n][columnId].(int)) case uint8: return int64(e.Rows[n][columnId].(uint8)) case uint16: return int64(e.Rows[n][columnId].(uint16)) case uint32: return int64(e.Rows[n][columnId].(uint32)) case uint64: return int64(e.Rows[n][columnId].(uint64)) case uint: return int64(e.Rows[n][columnId].(uint)) } return 0 } 

GetBinlogIdByName () पद्धति को छोड़कर सब कुछ तार्किक दिखता है।

इस तुच्छ सहायक को उनके सीरियल नंबरों के बजाय कॉलम नामों के साथ काम करने की आवश्यकता है, जो आपको निम्न की अनुमति देता है:

• गॉर्म एनोटेशन से कॉलम नाम लें;
• शुरुआत या मध्य में कॉलम जोड़ते समय संपादन करने की कोई आवश्यकता नहीं है;
• कॉलम नंबर 3 की तुलना में नाम क्षेत्र के साथ काम करना अधिक सुविधाजनक है।

परिणामस्वरूप, हम हैंडलर को स्वयं जोड़ते हैं:

 s.Field(k).SetInt(m.intHelper(e, n, columnName)) 

आइए दो और उदाहरण देखें।
ENUM: यहां मूल्य सूचकांक के रूप में आते हैं - अर्थात, "सक्रिय" स्थिति 1 के रूप में आएगी। ज्यादातर मामलों में, हमें एनम के एक स्ट्रिंग प्रतिनिधित्व की आवश्यकता है। इसे क्षेत्र विवरण से प्राप्त किया जा सकता है। एनम मूल्यों को पार्स करते समय, यह 1 से शुरू होता है, लेकिन संभावित मानों की सरणी स्वयं 0 से शुरू होती है।

एक Enum हैंडलर इस तरह लग सकता है:

 func (m *BinlogParser) stringHelper(e *canal.RowsEvent, n int, columnName string) string { columnId := m.getBinlogIdByName(e, columnName) if e.Table.Columns[columnId].Type == schema.TYPE_ENUM { values := e.Table.Columns[columnId].EnumValues //  if len(values) == 0 || e.Rows[n][columnId] == nil {{ return "" } return values[e.Rows[n][columnId].(int64)-1] //     0    } 

मैं JSON स्टोर करना चाहता हूं

अच्छा विचार क्यों नहीं। JSON mysql के संदर्भ में एक स्ट्रिंग है। किसी तरह यह इंगित करना आवश्यक है कि यह डेटा क्रमबद्ध है - इसके लिए हम गैर-वैज्ञानिक एनोटेशन "Json "से गॉर्म तक जोड़ देंगे।

कल्पना करें कि इस तरह की संरचना पर विचार किया जाना चाहिए:

 type JsonData struct { Int int `gorm:"column:int"` StructData TestData `gorm:"column:struct_data;fromJson"` MapData map[string]string `gorm:"column:map_data;fromJson"` SliceData []int `gorm:"column:slice_data;fromJson"` } type TestData struct { Test string `json:"test"` Int int `json:"int"` } 

आप बहुत सारी शर्तें लिख सकते हैं और शायद सफल हो सकते हैं। लेकिन प्रत्येक नया डेटा प्रकार सभी प्रयासों को मार देगा। यद्यपि स्टैकओवरफ़्लो के उत्तर खोजने का प्रयास - "कैसे एक अज्ञात प्रकार की संरचना को लाने और डिस्क्राइब करने के लिए" वाक्यांश के साथ शुरू होता है: "यह स्पष्ट नहीं है कि आपको इसकी आवश्यकता क्यों है, लेकिन कोशिश करें ..."।

इंटरफ़ेस को वांछित प्रकार देने के बाद, हम यह कर सकते हैं:

 if _, ok := parsedTag["FROMJSON"]; ok { newObject := reflect.New(s.Field(k).Type()).Interface() json := m.stringHelper(e, n, columnName) jsoniter.Unmarshal([]byte(json), &newObject) s.Field(k).Set(reflect.ValueOf(newObject).Elem().Convert(s.Field(k).Type())) } 

यदि आपके पास डेटा प्रकारों के बारे में प्रश्न हैं, तो आप परीक्षणों को देख सकते हैं या टिप्पणियों में पूछ सकते हैं।

आखिर में क्या हुआ ।