blackteay/open-im-server
1
0
Fork 0
mirror of https://github.com/openimsdk/open-im-server.git synced 2025-04-05 19:57:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: add kafka compress type and producer ack params (#1310)

Browse Source 
Signed-off-by: rfyiamcool <rfyiamcool@163.com>
This commit is contained in:
fengyun.rui 2023-11-10 22:04:16 +08:00 committed by GitHub
parent a580c15f6a
commit d1af343b13
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 24 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/common/config/config.go
View File

@ -85,10 +85,12 @@ type configStruct struct {
} `yaml:"redis"`
Kafka struct {
Username string `yaml:"username"`
Password string `yaml:"password"`
Addr []string `yaml:"addr"`
TLS *struct {
Username string `yaml:"username"`
Password string `yaml:"password"`
ProducerAck string `yaml:"producerAck"`
CompressType string `yaml:"compressType"`
Addr []string `yaml:"addr"`
TLS *struct {
CACrt string `yaml:"caCrt"`
ClientCrt string `yaml:"clientCrt"`
ClientKey string `yaml:"clientKey"`

19
pkg/common/kafka/producer.go
View File

@ -15,8 +15,10 @@
package kafka
import (
"bytes"
"context"
"errors"
"strings"
"time"
"github.com/OpenIMSDK/protocol/constant"
@ -49,8 +51,23 @@ func NewKafkaProducer(addr []string, topic string) *Producer {
p.config = sarama.NewConfig() // Instantiate a sarama Config
p.config.Producer.Return.Successes = true // Whether to enable the successes channel to be notified after the message is sent successfully
p.config.Producer.Return.Errors = true
p.config.Producer.RequiredAcks = sarama.WaitForAll // Set producer Message Reply level 0 1 all
p.config.Producer.Partitioner = sarama.NewHashPartitioner // Set the hash-key automatic hash partition. When sending a message, you must specify the key value of the message. If there is no key, the partition will be selected randomly
var producerAck = sarama.WaitForAll // default: WaitForAll
switch strings.ToLower(config.Config.Kafka.ProducerAck) {
case "no_response":
producerAck = sarama.NoResponse
case "wait_for_local":
producerAck = sarama.WaitForLocal
case "wait_for_all":
producerAck = sarama.WaitForAll
}
p.config.Producer.RequiredAcks = producerAck
var compress = sarama.CompressionNone // default: no compress
_ = compress.UnmarshalText(bytes.ToLower([]byte(config.Config.Kafka.CompressType)))
p.config.Producer.Compression = compress
if config.Config.Kafka.Username != "" && config.Config.Kafka.Password != "" {
p.config.Net.SASL.Enable = true
p.config.Net.SASL.User = config.Config.Kafka.Username