Vertx mongo客户端通过并发更新压倒mongodb

Question

尝试使用akka、rx、vert建立一个反应系统。x和mongodb中的流（下游）看起来很像：

<代码>发布者-

我遇到了这样一种情况，上游在mongo上调用了太多的更新收集操作，结果是：

<代码>com.mongodb.MongoWaitQueueFullException：等待连接的线程太多。已超过500个最大线程数（maxWaitQueueSize）

更新集合：

>

在同一集合上执行

正在向文档中嵌入的数组中添加新元素

1） 由于这是一个反应式系统，最好对源代码施加反压力，只执行尽可能频繁的http请求，因为这样可以写入mongo，这样等待连接mongo的线程队列就不会增长，

对于mongo的这种背压，有什么模式/例子可以遵循，或者我应该自己发明和实现它吗？

如何通过vertx mongo客户端访问和观察等待连接的线程数？

2） 在查看vertx mongo客户端的代码时，发现它没有保持与mongo的连接处于打开状态，并且每次更新操作都会打开新会话。您可以在io中观察到这一点。vertx。外景：mongo。实施。MongoClientImpl。updateCollection（…） 它下面调用MongoCollection的位置。updateOne（…）而不传递ClientSession ClientSession参数。

虽然每秒10次并发更新看起来很少，但问题是-可能是ClientSession的创建需要很长时间，因此导致线程排队的情况吗？

另外，在vertx mongo客户端中不缓存与mongo的连接的设计决策是什么？

mongostat是：

insert query update delete getmore command dirty  used flushes vsize   res qrw  arw net_in net_out conn                time
    *0    *0     32     *0       0   139|0  9.8% 10.8%       0 1.84G  666M 0|0 1|89  29.6k   62.9k  104 Jan 19 02:33:51.980
    *0    *0      6     *0       0     4|0 18.7% 18.7%       0 2.90G 1.50G 0|0 1|100  2.41k   9.59k  104 Jan 19 02:33:59.342
    *0    *0     *0     *0       0     2|0 15.7% 17.2%       0 3.52G 1.60G 0|0  1|97   493b   7.90k  104 Jan 19 02:34:07.480
    *0    *0      9     *0       0     3|0 14.7% 17.2%       0 3.52G 1.57G 0|0 1|100  3.10k   18.7k  104 Jan 19 02:34:10.955
    *0    *0      1     *0       0     1|0 21.4% 23.1%       0 3.52G 1.57G 0|0 1|100   749b   7.46k  104 Jan 19 02:34:19.579
    *0    *0     10     *0       0    16|0 36.7% 37.4%       0 3.57G 1.57G 0|0 1|100  4.79k   73.7k  104 Jan 19 02:34:20.443
    *0    *0     *0     *0       0     9|0 53.6% 54.0%       0 3.62G 1.56G 0|0 1|100  1.33k   47.6k  104 Jan 19 02:34:21.769
    *0    *0      1     *0       0    13|0 54.5% 55.2%       0 3.62G 1.57G 0|0 1|100  1.92k   70.6k  104 Jan 19 02:34:22.659
    *0    *0     *0     *0       0    23|0 70.5% 70.9%       0 3.62G 1.56G 0|0 1|100  2.75k    122k  104 Jan 19 02:34:23.173
    *0    *0     *0     *0       0    31|0 72.1% 72.5%       0 3.62G 1.58G 0|0 1|100  3.56k    153k  104 Jan 19 02:34:23.586

我非常感谢你的帮助。

Answer 1

我不认为你需要一个backpreassure，因为如果你一直得到的比你处理的更多，那么在你的情况下，它将分配所有内存和课程例外，但我认为选择是创建en emitter，其中每个流将项目发射到此发射器，并使用mongorepo。全部保存（发射器）：

进程1-

进程2-

进程3-

处理器4-

处理器5-

处理器6-

处理器7-

处理器8-

处理器9-

处理器10-