test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-01 11:29:48 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Refine insert when send message (#27491)

Browse Source 
Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>
Co-authored-by: Cai Zhang <cai.zhang@zilliz.com>
This commit is contained in:
zhenshan.cao 2023-10-18 21:12:08 +08:00 committed by GitHub
parent ac2d1bb5c2
commit 2b13078b14
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/mq/msgstream/mq_msgstream.go
View File

@ -56,7 +56,7 @@ type mqMsgStream struct {
closeRWMutex *sync.RWMutex
streamCancel func()
bufSize int64
producerLock *sync.Mutex
producerLock *sync.RWMutex
consumerLock *sync.Mutex
closed int32
onceChan sync.Once
@ -88,7 +88,7 @@ func NewMqMsgStream(ctx context.Context,
bufSize: bufSize,
receiveBuf: receiveBuf,
streamCancel: streamCancel,
producerLock: &sync.Mutex{},
producerLock: &sync.RWMutex{},
consumerLock: &sync.Mutex{},
closeRWMutex: &sync.RWMutex{},
closed: 0,
@ -288,13 +288,13 @@ func (ms *mqMsgStream) Produce(msgPack *MsgPack) error {
msg := &mqwrapper.ProducerMessage{Payload: m, Properties: map[string]string{}}
InjectCtx(spanCtx, msg.Properties)
ms.producerLock.Lock()
ms.producerLock.RLock()
if _, err := ms.producers[channel].Send(spanCtx, msg); err != nil {
ms.producerLock.Unlock()
ms.producerLock.RUnlock()
sp.RecordError(err)
return err
}
ms.producerLock.Unlock()
ms.producerLock.RUnlock()
}
}
return nil