记一次多线程数据拉取的网络优化思路

我们公司之前从京东慧采平台拉取的300W+条商品数据中,发现有大约78W条商品的marketPrice为空。
其中一部分是因为不在我们商品池,拉取不到价格数据;而另外一部分可能是最早拉取的代码有问题造成的,因为要得到marketPrice需要另外添加参数才能拉取到。
现在我需要对这78W条数据重新拉取价格信息。

首先分析该任务的流程:
1.从数据库获取待拉取的商品id。
2.从京东拉取价格信息。
3.价格回写入数据库。
很显然,这是一个IO密集型任务,其中网络请求消耗最大,而对CPU的计算依赖很小,所以我们可以稍微多启动一些线程来执行这次任务。
那么第一个问题,我们应该开多少个线程?
我想的是先开50个线程,然后看情况调整,但是旁边的同事说直接上100个线程,所以我们就先按照100个线程来试试水。

ExecutorService executor = new ThreadPoolExecutor(100, 100,
        0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(200),
        Executors.defaultThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy());

拒绝策略,使用调用者的线程执行,从而控制线程池队列的大小。
然后将查到的商品id,交给线程池去处理。
敲完代码,开跑。
很快我们就在控制台看到报错信息,果然Read TimeOut,而且还不少。
果然是线程起多了,大家都需要请求网络,而我的机器带宽还是被网管给限制了的,很容易就会TimeOut。
减少线程数是一个方法,但是以我的性格,肯定会再钻研一下。
首先我想到的是看网络请求情况,于是打开任务管理器。
这是100个线程直接进行网络请求的样子:

可以看到很明显的波峰和波谷。
再进行分析,由于分配任务给线程相对于整个周期来说,几乎是瞬时发生的,所以所有的线程其实都在同一时刻进行网络请求。
而网络请求大家的耗时也不会差太多,所以等大家都拿到数据,进行数据库更新时,又同时对数据库进行请求,造成了数据库压力。
(ps:这个库由于另外还有程序进行大量的增删改操作,本来压力就比较大了。)
如果能进行削峰,让网络请求和数据库请求都能更加平衡,就能提升整体的效率。
那么怎么削峰呢?
一听到削峰是不是有同学就想到MQ了?喂,你是不是网上培训老师的公开课听多了?
想到大学时好像是网络课还是什么课有讲过,在什么机制中,信道如果发生冲突,那么会随机睡眠一段时间,再发送,以解决信道冲突的问题。
好像是这样,但是具体也不记得了。
对于此处我们也可以使用相同的手段。
随机睡眠一段时间,让请求错峰。

Integer randomTime = new Random().nextInt(100);
try {
    Thread.sleep(randomTime);
} catch (InterruptedException e) {
    e.printStackTrace();
}

在主线程调度时,随机暂停0到100毫秒。
有朋友肯定会问,都sleep了,难道耗时不会变长?
那可不一定,理论上讲我们好像需要花更多的时间了,但其实随机睡眠0到200毫秒,这点时间对于整个生命周期的耗时来说,算不上什么,
不仅如此,因为减少了网络拥堵,效率更高了。
事实也证明,出现TimeOut的情况大大降低了。
我们先来拿出证据:

削峰的效果很明显,不用我说了,整体带宽消耗大概1.1Mbps。
但是中间出现了凹陷处,说明请求还有有些集中,所以我们修改下随机睡眠的参数,修改为0到200毫秒。

再看下优化后,注意纵坐标,之前最大纵坐标是11Mbps,现在是1Mbps,优化后的带宽占用甚至不到800Kbps。

ok,根据之前峰值的情况来看,我电脑的带宽还是有大量压榨的空间,所以100个线程可能还有点少。所以我加大到200个线程。
但是根据数据库查询的待更新商品数量来看,效果提升并不大。
于是暴力一点,增加到400个线程,结果发现,100、200、400个线程,每分钟的处理速度差不多。
这就奇怪了,是哪儿拖了后腿呢?
看了下打印的日志,发现线程结束后没有马上进行下一项任务,难道队列还没有塞满?
即使我们都按照睡眠200毫秒,那么也应该会塞满啊。
然后我想到了查询商品id的sql,我们在marketPrice字段上没有做索引,因为平时不会以这个字段来查询,所以我的sql是会走全表扫描的,再加上数据库压力较大,所以一个sql竟然要惊人的30秒,难怪队列都塞不满。
于是我减小随机睡眠时间,增加了单次查询的数量,同时增大阻塞队列大小,让主线程提前加载批待处理的商品id。
Bingo,确实是怎样,效果很明显。
对此,虽然我们不太好优化sql,但是我们可能加大阻塞队列的大小,让主线程提前加载下一批待处理的商品id,提高利用率。
接下来就是调整线程大小,阻塞队列大小和随机睡眠时间参数,让处理速度尽可能得快。
所以说,并不是线程越多执行的越快,而是应该对任务进行分析,根据实际遇到的瓶颈来进行调整,才能得到一个较好的结果。

发表评论

电子邮件地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据