请教一个并发设计问题

[swqslwl]

我在做一个监测流量的项目。每秒会从数据源中获取 1w 条 json 格式的流量信息，我希望对这些流量进行分析，但是现在会出现丢数据的情况。

我的做法是
1.接受到数据后先传入 channelA
2.启动一个协程循环从 A 中读取数据存入切片 B
3.另起一个协程处理切片 B 的数据，同时在处理业务时利用 mutex 锁住 B

实际调试中发现，mutex 的次数会影响数据的丢失量
请问我这样设计是否有问题，是否会导致丢数据

[RH]

需要 demo 才能分析，你描述的逻辑里有很多不确定性。

[lrh3321]

要 demo, 你两个协程都会写切片 B

[ox180]

@lrh3321 冒泡

[piaodazhu]

我就提一个可能，切片 B 扩容，导致这种特殊情况：
时刻 1 ，goroutine1 加锁，用 B=append(B, item)向切片 B 追加一个元素。刚好触发了扩容，B 的底层数组指针发生了转移。即，append 的参数 B 和返回值 B 中的 ptr 不同。
时刻 2 紧接着时刻 1 ，goroutine2 拿到锁，这个时候在 goroutine2 看来，B 只是一个由(size,cap,ptr)构成结构体，它察觉不到 B 底层数组指针的变化，所以看不到 goroutine1 追加的数据。

具体可以检查一下代码。

[xuboying]

我感觉 sync.Pool 是干这个事情的。但是我一直没有掌握 sync.Pool 的正确用法，希望有大佬解释一下。

[ding2dong]

调大 A 的 bufsize

[ding2dong]

另外写入 B 的时候也要 mutex ，否则会被污染

[lysS]

“但是现在会出现丢数据的情况” 这是为什么呢？实际没有从数据源中获取到 1w 条？

[liangkang1436]

有没有考虑用时序数据库来存储这些数据然后订阅？ 1w/s ，这个数据量不小了

[fregie]

2 中存入 b 的过程也要锁。
其实这里不用切片用队列比较合适

[ghost024]

没看到你的代码，粗略的分析，你的第一个协程，从 channelA 中写到切片 b 也需要先获得 b 的 mutex 锁的，要不然，如果在锁 b 的时候你从 channelA 中获取数据，因为 b 锁住了，你写不进去就丢了

[WispZhan]

The Golden Rule - Don't Block the Event Loop or Coroutine.

[Martens]

写切片 B 和读切片 B 的时候都要加锁

[dode]

先放 kafka ，再批量读出来处理呢

[8355]

@ghost024 #11 +1 跟我理解的一样
而且整个过程感觉效率并不高
使用中间件哪怕 redis stream 整个代码都可以简单很多

[joesonw]

能确定数量就用 channel ，不行的话用 linked list 。尽量避免用锁，传递锁的时候要传指针&。

[matrix1010]

丢数据算 bug 吗? 如果算请写个并发的单元测试并加上-race 测一下

[Nazz]

数据量有点大, 建议使用 sync.Pool + 任务队列

[swqslwl]

@lrh3321
@RH 老哥代码放上了

[swqslwl]

@ding2dong
@fregie

@ghost024
@Martens
@8355 对，这里确实是有问题。但是我加上后发现还是会丢