From d41f2a7aef2802f9656241873b25d23ff391d30a Mon Sep 17 00:00:00 2001 From: plutoyty <2631223275@qq.com> Date: Wed, 12 Jul 2023 20:21:52 +0800 Subject: [PATCH] Add retry mechanism to mongoDB, Redis, Kafka --- config/config.yaml | 59 +++++++++++++++++--------- pkg/common/db/cache/init_redis_test.go | 6 +++ pkg/common/kafka/producer.go | 3 ++ 3 files changed, 47 insertions(+), 21 deletions(-) diff --git a/config/config.yaml b/config/config.yaml index a001c68c5..4ff0e44b7 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -56,11 +56,14 @@ kafka: topic: "offlineMsgToMongoMysql" #不建议修改 msgToPush: topic: "msgToPush" #不建议修改 + msgToModify: + topic: "msgToModify" #不建议修改 consumerGroupID: #消费者组，不建议修改 msgToRedis: redis # msgToMongo: mongo # msgToMySql: mysql # msgToPush: push # + msgToModify: modify # rpc: @@ -73,26 +76,41 @@ api: listenIP: #默认为0.0.0.0 object: - enable: "minio" #使用minio - apiURL: "http://127.0.0.1:10002/object/" + enable: minio #使用minio + apiURL: http://127.0.0.1:10002/third/object minio: - bucket: "openim" #不建议修改 - endpoint: "http://127.0.0.1:10005" #minio对外服务的ip和端口，app要能访问此ip和端口 - accessKeyID: "root" #ID - secretAccessKey: "openIM123" #秘钥 - sessionToken: "" #token - cos: #tencent cos - bucketURL: "https://temp-1252357374.cos.ap-chengdu.myqcloud.com" - secretID: "" - secretKey: "" - sessionToken: "" - oss: #ali oss - endpoint: "https://oss-cn-chengdu.aliyuncs.com" - bucket: "demo-9999999" - bucketURL: "https://demo-9999999.oss-cn-chengdu.aliyuncs.com" - accessKeyID: "" - accessKeySecret: "" - sessionToken: "" + tempBucket: "openim" #不建议修改 + dataBucket: "openim" #不建议修改 + location: us-east-1 #不建议修改 + endpoint: http://127.0.0.1:10005 #minio对外服务的ip和端口，app要能访问此ip和端口 + accessKeyID: root #ID + secretAccessKey: openIM123 #秘钥 + isDistributedMod: false #是否分布式多硬盘部署，如果是多硬盘部署，需要修改为true + tencent: #tencent cos + appID: + region: + bucket: + secretID: + secretKey: + ali: #ali oss + regionID: + accessKeyID: + accessKeySecret: + stsEndpoint: + ossEndpoint: + bucket: + finalHost: + stsDurationSeconds: + OssRoleArn: + aws: + accessKeyID: + accessKeySecret: + region: + bucket: + finalHost: + roleArn: + externalId: + roleSessionName: rpcPort: #rpc服务端口，不建议修改，端口由脚本读取后传入程序，如启动多个程序，只需要填入多个端口，用逗号隔开，如 [10110, 10111] openImUserPort: [ 10110 ] @@ -164,8 +182,7 @@ groupMessageHasReadReceiptEnable: true #群聊已读是否开 singleMessageHasReadReceiptEnable: true #单聊已读是否开启 retainChatRecords: 365 #mongo保存离线消息时间（天） -chatRecordsClearTime: "0 2 * * 3" #每周三凌晨2点清理mongo中的过期（超过retainChatRecords时间）消息，这个删除是为了清理满足上个配置retainChatRecords的过期消息，不会发送通知，仅仅作为清理磁盘使用 -msgDestructTime: "0 2 * * *" #消息自动删除时间，每天凌晨2点删除过期消息，这个删除是为了删除保留时间超过超过会话字段msg_destruct_time（秒）的消息。 +chatRecordsClearTime: "0 2 * * 3" #每周三凌晨2点清理mongo中的过期（超过retainChatRecords时间）消息 secret: tuoyun #秘钥，获取token时校验 diff --git a/pkg/common/db/cache/init_redis_test.go b/pkg/common/db/cache/init_redis_test.go index 6f78a43bd..7bf1a4a7d 100644 --- a/pkg/common/db/cache/init_redis_test.go +++ b/pkg/common/db/cache/init_redis_test.go @@ -27,4 +27,10 @@ func TestNewRedis(t *testing.T) { fmt.Println("config load error") return } + redis, err := NewRedis() + if err != nil { + fmt.Println(err) + return + } + fmt.Println(redis) } diff --git a/pkg/common/kafka/producer.go b/pkg/common/kafka/producer.go index a66ef3dba..a749c76f8 100644 --- a/pkg/common/kafka/producer.go +++ b/pkg/common/kafka/producer.go @@ -76,6 +76,9 @@ func NewKafkaProducer(addr []string, topic string) *Producer { //} time.Sleep(time.Duration(1) * time.Second) } + if err != nil { + panic(err.Error()) + } p.producer = producer return &p }