Java 如何实现异步更新数据库呢？

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

这是一个创建于 2614 天前的主题，其中的信息可能已经有所发展或是发生改变。

如题：

比如程序要执行以下步骤：

A：往单例类的 list 属性中写数据（ list 的 size 到达 100 的时候，把 list 保存到数据中，同时清空 list ）

B：执行业务，return true

现在希望是程序进来的时候，马上执行 B，异步去执行 A，请问如何实现呢？

谢谢大神们！

list

异步

执行

单例类

27 条回复 • 2017-10-13 18:08:31 +08:00

tinyuu

2017-10-07 11:02:44 +08:00

线程啊

chunrong918

2017-10-07 11:06:17 +08:00

@tinyuu

你的意思是每次进来的时候 new 一个 thread，去执行 A 操作?

用同步锁锁住单例累，执行 A ？

有没有办法用并发包？

hwding

2017-10-07 11:08:34 +08:00 via iPhone

带优先级的阻塞队列？就像消息那样。

chunrong918

2017-10-07 11:09:48 +08:00

@tinyuu 往 list 属性中写和清空，应该算一个 pv 操作？

chunrong918

2017-10-07 11:11:11 +08:00

@hwding 有参考资料吗？能大概讲一下吗？因为我以前没接触过

hwding

2017-10-07 11:16:16 +08:00 via iPhone

@chunrong918 大概是多线程和并发以及异步这一块的吧，具体哪本书我也说不上来。也可以选用一些库。

BBCCBB

2017-10-07 11:24:18 +08:00

要用并发库就把 list 改为 blockingqueue 或者 concurrentqueue 之类的，然后就是线程了

v2orz

2017-10-07 11:28:11 +08:00

不要用单例类的 list 而是用队列
往线程池提交任务也行，其实也是内部的队列

azygote

2017-10-07 11:29:57 +08:00 via iPhone

用 redis 队列

hantsy

2017-10-07 11:38:55 +08:00

用 Spring 一个 @Async 就可以了。
更好的方法用一些 Messaging Broker，如 RabbitMQ 去处理。

tinyuu

2017-10-07 11:42:33 +08:00

@chunrong918 开始就创建一个线程，不要用 list。用 queue ；

chocotan

2017-10-07 11:56:01 +08:00

貌似可以用 rxjava 的 buffer

sagaxu

2017-10-07 13:55:07 +08:00

在初始化 bean 的时候创建一个 ConcurrentLinkedQueue 和一个 AtomicInteger 计数器，还要注入一个 ThreadPoolExecutor。

A 往 queue 里塞数据的时候计数器加 1，如果满 100，把计数器减 100，并且往 ThreadPoolExecutor 里扔一个消费 100 个数据的 task，这里减计数器和扔 task 要做好同步，用 double check 加锁简单同步下就可以了

if (counter >= 100) {
synchronized (this) {
if (counter >= 100) {
}
}
}

除了每满 100 个，还可以控制下时间，比如上一次写入 db 有 5 分钟了，那么不管当前满没满 100 也要把 queue 里的数据写入 db，光靠数量控制是不够的。

需要特别注意的是，不要在 controller 里创建线程，那是个不好的习惯，容器里的线程是被托管的，你在托管的线程里创建自己的线程，会带来潜在问题。

yidinghe

2017-10-07 14:12:44 +08:00 via Android

1、肯定是用另外的线程；
2、如果并发量很大，就用线程池来控制负载。

loveCoding

2017-10-07 14:28:13 +08:00

队列+线程池吧

movistar

2017-10-07 15:04:00 +08:00

经典的生产者消费者模型
另外是不是真的到 100 个要 flush 一次呢,这个是业务场景?
一般来说异步模式的消费者会这么做:
提供一个最长 flush time,以及一个队列最长长度,如果到这个时间阈值,队列还没满,那么直接清空队列进行消费
如果还没到时间阈值,队列满了,就直接清空队列进行消费
需要考虑 13 楼说的,消费者过慢的问题,提供多个消费者,在多个线程中进行消费,避免 block
有很多实现方式,可以参考一下 ElasticSearch 的 BulkProcessor
相对比较完善比较完整,如果不需要这么复杂的逻辑,可以简化一下
https://github.com/elastic/elasticsearch/blob/c47f24d4061f51c8e831d030443afa90d73f681c/core/src/main/java/org/elasticsearch/action/bulk/BulkRequestHandler.java

sorra

2017-10-07 17:45:05 +08:00

在 Executors 里选一种线程池，往里面提交任务就可以，注意正确选择队列排满时的行为(阻塞或拒绝)。
生产环境推荐用专门的消息队列来做，不会丢失任务。

boywang004

2017-10-07 19:15:23 +08:00

https://github.com/PhantomThief/buffer-trigger
之前抽象过一个这样的组件，不过是 Java8 only 的，仅供参考和借鉴。

sudoz

2017-10-07 21:04:45 +08:00

线程池 MQ

chunrong918

2017-10-07 23:20:28 +08:00

@hwding 这边不需要考虑优先级，感觉阻塞队列加线程就可以解决了

chunrong918

2017-10-07 23:21:05 +08:00

@BBCCBB 是啊，阻塞队列加线程就可以解决了

chunrong918

2017-10-07 23:52:04 +08:00

@movistar 确实向你说的，考虑时间和队列长度问题，但是 elasticsearch 的那个类源码看不太懂，有点复杂呢

chunrong918

2017-10-08 00:06:29 +08:00

@sagaxu 好办法，这边的控制时间，定时去更新是使用 java 的定时器 timer 还是 crontab 啊？

chunrong918

2017-10-08 00:07:09 +08:00

@sagaxu 有相关代码可以参考一下吗

codeyung

2017-10-08 00:09:04 +08:00 via iPhone

异步或者队列线程可以看下相关文档和 GitHub

cxbig

2017-10-08 00:42:39 +08:00

一般我们都是用队列（ Queue ）做平衡

upupxjg

2017-10-13 18:08:31 +08:00

多线程，不是说 A、B 搞两个线程，而是把 A 写库这个事儿扔到线程池里去，List 攒够 100 （必须是 100 么？多一点都不行？） A 加锁，把这个 List 扔给线程池，再 new 一个 list 解锁。打完收工