sql同步消息添加id,处理消息重复的问题

zhangzengfei

2023-11-08 2ab94056786f8a7bcf66cf1ed25002e74ed7f016

sql同步消息添加id,处理消息重复的问题

 serf/sync.go

@@ -1,11 +1,9 @@
package serf

import (
    "apsClient/pkg/logx"
    "context"
    "encoding/json"
    "fmt"
    "github.com/mitchellh/mapstructure"
    "os"
    "os/signal"
    "regexp"
@@ -13,12 +11,17 @@
    "syscall"
    "time"

    "apsClient/pkg/logx"

    "basic.com/pubsub/protomsg.git"
    "basic.com/valib/bhomeclient.git"
    "basic.com/valib/bhomedbapi.git"

    "github.com/gogo/protobuf/proto"
    "github.com/jinzhu/gorm"
    "github.com/satori/go.uuid"
    "github.com/mitchellh/mapstructure"
    "github.com/muesli/cache2go"
)

var (
@@ -26,6 +29,8 @@
    dependProcs = []string{
        bhomeclient.Proc_System_Service,
    }

    sqlMsgSeqCache = cache2go.Cache("syncSqlMsg")
)

const (
@@ -44,6 +49,12 @@
    Proc    string `json:"procName"` // 进程名
    Topic   string `json:"topic"`    // 主题
    Payload []byte `json:"payload"`  // 消息体,自行解析
}

type SqlMsg struct {
    Id      string
    Sql     string
    Version string
}

type SyncServer struct {
@@ -169,13 +180,20 @@
    os.Exit(0)
}

func (ss *SyncServer) pubSyncSqlMessage(payload []byte, targetId string) error {
func (ss *SyncServer) pubSyncSqlMessage(sql string, targetId string) error {
    sqlMsg := SqlMsg{
        Id:  uuid.NewV4().String(),
        Sql: sql,
    }

    bMsg, _ := json.Marshal(sqlMsg)

    var msg = ProcMessageEvent{
        Owner:   ss.ServerId,
        Target:  targetId,
        Proc:    ss.ProcName,
        Topic:   ss.syncSqlTopic,
        Payload: payload,
        Payload: bMsg,
    }

    b, err := json.Marshal(msg)
@@ -344,7 +362,7 @@
                syncSql := strings.Join(sqlBuf, "")

                //fmt.Println("同步sql语句:", syncSql)
                ss.pubSyncSqlMessage([]byte(syncSql), "")
                ss.pubSyncSqlMessage(syncSql, "")

                sqlBuf = append([]string{})
                sendSize = 0
@@ -355,7 +373,7 @@
                    syncSql := strings.Join(sqlBuf, "")
                    //fmt.Println("同步sql语句:", syncSql)

                    ss.pubSyncSqlMessage([]byte(syncSql), "")
                    ss.pubSyncSqlMessage(syncSql, "")

                    sqlBuf = append([]string{})
                }
@@ -373,9 +391,25 @@
    }
}

func (ss *SyncServer) handleClusterMessage(msg []byte) {
    logx.Infof("clusterMessage:", string(msg))
    sql := string(msg)
func (ss *SyncServer) handleClusterMessage(clusterMsgData []byte) {
    var msg SqlMsg
    err := json.Unmarshal(clusterMsgData,&msg)
    if err != nil {
        logx.Errorf(" Unmarshal cluster message error, %s",err.Error())
        return
    }

    // 判断消息是否曾经接收过
    if sqlMsgSeqCache.Exists(msg.Id) {
        logx.Infof("clusterMessage:接收到重复消息, %s", msg.Sql)
        return
    }

    // 记录消息id, 半小时过期
    sqlMsgSeqCache.Add(msg.Id, 30*time.Minute, true)

    logx.Infof("clusterMessage:%s", msg.Sql)
    sql := msg.Sql

    if len(sql) <= 0 {
        return
@@ -420,12 +454,12 @@
    logx.Infof("DumpTables sql:%v", sqls)
    syncSql := strings.Join(sqls, ";")
    if len(syncSql) < sizeLimit {
        err = ss.pubSyncSqlMessage([]byte(syncSql), targetId)
        err = ss.pubSyncSqlMessage(syncSql, targetId)
    } else {
        shard := ""
        for _, sql := range sqls {
            if len(shard)+len(sql) > sizeLimit {
                err = ss.pubSyncSqlMessage([]byte(shard), targetId)
                err = ss.pubSyncSqlMessage(shard, targetId)
                shard = ""
            }

@@ -433,7 +467,7 @@
        }

        if len(shard) > 0 {
            err = ss.pubSyncSqlMessage([]byte(shard), targetId)
            err = ss.pubSyncSqlMessage(shard, targetId)
        }
    }