犀利豆的博客

犀利豆的博客

马上订阅 犀利豆的博客 RSS 更新: https://xilidou.com/atom.xml

徒手撸框架--高并发环境下的请求合并

2018年1月23日 03:10

原文地址：https://www.xilidou.com/2018/01/22/merge-request/

在高并发系统中，我们经常遇到这样的需求：系统产生大量的请求，但是这些请求实时性要求不高。我们就可以将这些请求合并，达到一定数量我们统一提交。最大化的利用系统性IO,提升系统的吞吐性能。

所以请求合并框架需要考虑以下两个需求：

  1. 当请求收集到一定数量时提交数据
  2. 一段时间后如果请求没有达到指定的数量也进行提交

我们就聊聊一如何实现这样一个需求。

阅读这篇文章你将会了解到:

  • ScheduledThreadPoolExecutor
  • 阻塞队列
  • 线程安全的参数
  • LockSupport的使用

设计思路和实现

我们就聊一聊实现这个东西的具体思路是什么。希望大家能够学习到分析问题，设计模块的一些套路。

1. 底层使用什么数据结构来持有需要合并的请求？

  • 既然我们的系统是在高并发的环境下使用，那我们肯定不能使用，普通的ArrayList来持有。我们可以使用阻塞队列来持有需要合并的请求。
  • 我们的数据结构需要提供一个 add() 的方法给外部，用于提交数据。当外部add数据以后，需要检查队列里面的数据的个数是否达到我们限额？达到数量提交数据，不达到继续等待。
  • 数据结构还需要提供一个timeOut()的方法，外部有一个计时器定时调用这个timeOut方法，如果方法被调用，则直接向远程提交数据。
  • 条件满足的时候线程执行提交动作，条件不满足的时候线程应当暂停，等待队列达到提交数据的条件。所以我们可以考虑使用 LockSupport.park()和LockSupport.unpark 来暂停和激活操作线程。

经过上面的分析，我们就有了这样一个数据结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
private static class FlushThread<Item> implements Runnable{

        private final String name;

        //队列大小
        private final int bufferSize;
        //操作间隔
        private int flushInterval;

        //上一次提交的时间。
        private volatile long lastFlushTime;
        private volatile Thread writer;

        //持有数据的阻塞队列
        private final BlockingQueue<Item> queue;

        //达成条件后具体执行的方法
        private final Processor<Item> processor;

        //构造函数
        public FlushThread(String name, int bufferSize, int flushInterval,int queueSize,Processor<Item> processor) {
            this.name = name;
            this.bufferSize = bufferSize;
            this.flushInterval = flushInterval;
            this.lastFlushTime = System.currentTimeMillis();
            this.processor = processor;

            this.queue = new ArrayBlockingQueue...
剩余内容已隐藏
查看完整文章以阅读更多
查看完整文章