资讯专栏INFORMATION COLUMN

Java多线程编程实战:模拟大量数据同步

elliott_hu / 2185人阅读

摘要:所以得出结论需要分配较多的线程进行读数据,较少的线程进行写数据。注意多线程编程对实际环境和需求有很大的依赖,需要根据实际的需求情况对各个参数做调整。

背景

最近对于 Java 多线程做了一段时间的学习,笔者一直认为,学习东西就是要应用到实际的业务需求中的。否则要么无法深入理解,要么硬生生地套用技术只是达到炫技的效果。

不过笔者仍旧认为自己对于多线程掌握不够熟练,不敢轻易应用到生产代码中。这就按照平时工作中遇到的实际问题,脑补了一个很可能存在的业务场景:

已知某公司管理着 1000 个微信服务号,每个服务号有 1w ~ 50w 粉丝不等。假设该公司每天都需要将所有微信服务号的粉丝数据通过调用微信 API 的方式更新到本地数据库。

需求分析

对此需求进行分析,主要存在以下问题:

单个服务号获取粉丝 id,只能每次 1w 按顺序拉取

微信的 API 对于服务商的并发请求数量有限制

单个服务号获取粉丝 id,只能每次 1w 按顺序拉取。这个问题决定了单个公众号在拉取粉丝 id 上,无法分配给多个线程执行。

微信的 API 对于服务商的并发请求数量有限制。这点最容易被忽略,如果我们同时有过多的请求,则会导致接口被封禁。这里可以通过信号量来控制同时执行的线程数量。

为了尽快完成数据同步,根据实际情况:整个数据同步可分为读数据和写数据两个部分。读数据是通过 API 获取,走网络 IO,速度较慢;写数据是写到数据库,速度较快。所以得出结论:需要分配较多的线程进行读数据,较少的线程进行写数据。

设计要点

首先,我们需要确定开启多少个线程(在生产中往往是使用线程池),线程数量需要根据服务器性能来决定,这里我们定为 40 个读取数据线程(将 1000 个公众号分为 40 份,分别在 40 个线程中执行),1个写入数据线程。(具体开多少个线程,取决于线程池的容量,以及可以分配给此业务的数量。具体的数字需要根据实际情况测试得出,比服务器阈值低一些较好。当然,配置允许范围内越大越好)

其次,考虑到微信对于 API 并发请求的限制,需要限制同时执行的线程数,使用java.util.concurrent.Semaphore进行控制,这里我们限制为 20 个(具体的信号量凭证数,取决于同一时间能够执行的线程,跟 API 限制,服务器性能有关)。

然后,我们需要知道数据何时读取、写入完毕,以控制程序逻辑以及终止程序,这里我们使用java.util.concurrent.CountDownLatch进行控制。

最后,我们需要一个数据结构,用来在多个线程中共享处理的数据,此处同步数据的场景非常适合使用队列,这里我们使用线程安全的java.util.concurrent.ConcurrentLinkedQueue来进行处理。(需要注意的是,在实际开发中,队列不能够无限制地增长,这将会很快消耗掉内存,我们需要根据实际情况对队列长度做控制。例如,可以通过控制读取线程数和写入线程数的比例来控制队列的长度)

模拟代码

由于本文重点关注多线程的使用,模拟代码只体现多线程操作的方法。代码里添加了大量的注释,方便各位读者阅读理解。

JDK:1.8

import java.util.Arrays;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * N个线程向队列添加数据
 * 一个线程消费队列数据
 */
public class QueueTest {
    private static List data = Arrays.asList("a", "b", "c", "d", "e");

    private static final int OFFER_COUNT = 40; // 开启的线程数量

    private static Semaphore semaphore = new Semaphore(20); // 同一时间执行的线程数量(大多用于控制API调用次数或数据库查询连接数)

    public static void main(String[] args) throws InterruptedException {
        Queue queue = new ConcurrentLinkedQueue<>(); // 处理队列,需要处理的数据,放置到此队列中

        CountDownLatch offerLatch = new CountDownLatch(OFFER_COUNT); // offer线程latch,每完成一个,latch减一,lacth的count为0时表示offer处理完毕
        CountDownLatch pollLatch = new CountDownLatch(1); // poll线程latch,latch的count为0时,表示poll处理完毕

        Runnable offerRunnable = () -> {
            try {
                semaphore.acquire(); // 信号量控制
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            try {
                for (String datum : data) {
                    queue.offer(datum);
                    TimeUnit.SECONDS.sleep(2); // 模拟取数据很慢的情况
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 在finally中执行latch.countDown()以及信号量释放,避免因异常导致没有正常释放
                offerLatch.countDown();
                semaphore.release();
            }
        };

        Runnable pollRunnable = () -> {
            int count = 0;
            try {
                while (offerLatch.getCount() > 0 || queue.size() > 0) { // 只要offer的latch未执行完,或queue仍旧有数据,则继续循环
                    String poll = queue.poll();
                    if (poll != null) {
                        System.out.println(poll);
                        count++;
                    }
                    // 无论是否poll到数据,均暂停一小段时间,可降低CPU消耗
                    TimeUnit.MILLISECONDS.sleep(100);
                }
                System.out.println("total count:" + count);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                // 在finally中执行latch.countDown(),避免因异常导致没有正常释放
                pollLatch.countDown();
            }
        };

        // 启动线程(生产环境中建议使用线程池)
        new Thread(pollRunnable).start(); // 启动一个poll线程
        for (int i = 0; i < OFFER_COUNT; i++) {
            new Thread(offerRunnable).start();
        } // 模拟取数据很慢,需要开启40个线程处理

        // latch等待,会block主线程直到latch的count为0
        offerLatch.await();
        pollLatch.await();

        System.out.println("===the end===");
    }
}

到这里,本文结束。以上是笔者脑补的一个常见需求的解决方案。

注意:多线程编程对实际环境和需求有很大的依赖,需要根据实际的需求情况对各个参数做调整。实际在使用中,需要尽量模拟生产环境的数据情况来进行测试,对服务器执行期间的并发数,CPU、内存、网络 IO、磁盘 IO 做好观察。并适当地调低并发数,以给服务器留有处理其他请求的余量。

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/73249.html

相关文章

  • 后端好书阅读与推荐(续三)

    摘要:后端好书阅读与推荐系列文章后端好书阅读与推荐后端好书阅读与推荐续后端好书阅读与推荐续二后端好书阅读与推荐续三这里依然记录一下每本书的亮点与自己读书心得和体会,分享并求拍砖。然后又请求封锁,当释放了上的封锁之后,系统又批准了的请求一直等待。 后端好书阅读与推荐系列文章:后端好书阅读与推荐后端好书阅读与推荐(续)后端好书阅读与推荐(续二)后端好书阅读与推荐(续三) 这里依然记录一下每本书的...

    ckllj 评论0 收藏0
  • 后端好书阅读与推荐(续三)

    摘要:后端好书阅读与推荐系列文章后端好书阅读与推荐后端好书阅读与推荐续后端好书阅读与推荐续二后端好书阅读与推荐续三这里依然记录一下每本书的亮点与自己读书心得和体会,分享并求拍砖。然后又请求封锁,当释放了上的封锁之后,系统又批准了的请求一直等待。 后端好书阅读与推荐系列文章:后端好书阅读与推荐后端好书阅读与推荐(续)后端好书阅读与推荐(续二)后端好书阅读与推荐(续三) 这里依然记录一下每本书的...

    jcc 评论0 收藏0
  • 后端好书阅读与推荐(续三)

    摘要:后端好书阅读与推荐系列文章后端好书阅读与推荐后端好书阅读与推荐续后端好书阅读与推荐续二后端好书阅读与推荐续三这里依然记录一下每本书的亮点与自己读书心得和体会,分享并求拍砖。然后又请求封锁,当释放了上的封锁之后,系统又批准了的请求一直等待。 后端好书阅读与推荐系列文章:后端好书阅读与推荐后端好书阅读与推荐(续)后端好书阅读与推荐(续二)后端好书阅读与推荐(续三) 这里依然记录一下每本书的...

    lauren_liuling 评论0 收藏0
  • Java线程学习(七)并发编程中一些问题

    摘要:相比与其他操作系统包括其他类系统有很多的优点,其中有一项就是,其上下文切换和模式切换的时间消耗非常少。因为多线程竞争锁时会引起上下文切换。减少线程的使用。很多编程语言中都有协程。所以如何避免死锁的产生,在我们使用并发编程时至关重要。 系列文章传送门: Java多线程学习(一)Java多线程入门 Java多线程学习(二)synchronized关键字(1) java多线程学习(二)syn...

    dingding199389 评论0 收藏0

发表评论

0条评论

最新活动
阅读需要支付1元查看
<