使用双异步后,如何保证数据一致性?

这篇具有很好参考价值的文章主要介绍了使用双异步后,如何保证数据一致性?。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

使用双异步后,如何保证数据一致性?,搬砖工逆袭Java架构师,java,高并发,异步,线程池,学习

大家好,我是哪吒。

一、前情提要

在上一篇文章中,我们通过双异步的方式导入了10万行的Excel,有个小伙伴在评论区问我,如何保证插入后数据的一致性呢?

很简单,通过对比Excel文件行数和入库数量是否相等即可。

那么,如何获取异步线程的返回值呢?

使用双异步后,如何保证数据一致性?,搬砖工逆袭Java架构师,java,高并发,异步,线程池,学习

二、通过Future获取异步返回值

我们可以通过给异步方法添加Future返回值的方式获取结果。

FutureTask 除了实现 Future 接口外,还实现了 Runnable 接口。因此,FutureTask 可以交给 Executor 执行,也可以由调用线程直接执行FutureTask.run()。

1、FutureTask 是基于 AbstractQueuedSynchronizer实现的

AbstractQueuedSynchronizer简称AQS,它是一个同步框架,它提供通用机制来原子性管理同步状态、阻塞和唤醒线程,以及 维护被阻塞线程的队列。
基于 AQS 实现的同步器包括: ReentrantLock、Semaphore、ReentrantReadWriteLock、 CountDownLatch 和 FutureTask。

基于 AQS实现的同步器包含两种操作:

  1. acquire,阻塞调用线程,直到AQS的状态允许这个线程继续执行,在FutureTask中,get()就是这个方法;
  2. release,改变AQS的状态,使state变为非阻塞状态,在FutureTask中,可以通过run()和cancel()实现。

2、FutureTask执行流程

使用双异步后,如何保证数据一致性?,搬砖工逆袭Java架构师,java,高并发,异步,线程池,学习

  1. 执行@Async异步方法;
  2. 建立新线程async-executor-X,执行Runnable的run()方法,(FutureTask实现RunnableFuture,RunnableFuture实现Runnable);
  3. 判断状态state;
    • 如果未新建或者不处于AQS,直接返回;
    • 否则进入COMPLETING状态,执行异步线程代码;
  4. 如果执行cancel()方法改变AQS的状态时,会唤醒AQS等待队列中的第一个线程线程async-executor-1;
  5. 线程async-executor-1被唤醒后
    • 将自己从AQS队列中移除;
    • 然后唤醒next线程async-executor-2;
    • 改变线程async-executor-1的state;
    • 等待get()线程取值。
  6. next等待线程被唤醒后,循环线程async-executor-1的步骤
    • 被唤醒
    • 从AQS队列中移除
    • 唤醒next线程
    • 改变异步线程状态
  7. 新建线程async-executor-N,监听异步方法的state
    • 如果处于EXCEPTIONAL以上状态,抛出异常;
    • 如果处于COMPLETING状态,加入AQS队列等待;
    • 如果处于NORMAL状态,返回结果;

3、get()方法执行流程

get()方法通过判断状态state观测异步线程是否已结束,如果结束直接将结果返回,否则会将等待节点扔进等待队列自旋,阻塞住线程。

自旋直至异步线程执行完毕,获取另一边的线程计算出结果或取消后,将等待队列里的所有节点依次唤醒并移除队列。

使用双异步后,如何保证数据一致性?,搬砖工逆袭Java架构师,java,高并发,异步,线程池,学习

  1. 如果state小于等于COMPLETING,表示任务还在执行中;
    • 如果线程被中断,从等待队列中移除等待节点WaitNode,抛出中断异常;
    • 如果state大于COMPLETING;
      • 如果已有等待节点WaitNode,将线程置空;
      • 返回当前状态;
    • 如果任务正在执行,让出时间片;
    • 如果还未构造等待节点,则new一个新的等待节点;
    • 如果未入队列,CAS尝试入队;
    • 如果有超时时间参数;
      • 计算超时时间;
      • 如果超时,则从等待队列中移除等待节点WaitNode,返回当前状态state;
      • 阻塞队列nanos毫秒。
    • 否则阻塞队列;
  2. 如果state大于COMPLETING;
    • 如果执行完毕,返回结果;
    • 如果大于等于取消状态,则抛出异常。

很多小朋友对读源码,嗤之以鼻,工作3年、5年,还是没认真读过任何源码,觉得读了也没啥用,或者读了也看不懂~

其实,只要把源码的执行流程通过画图的形式呈现出来,你就会幡然醒悟,原来是这样的~

简而言之:

1. 如果异步线程还没执行完,则进入CAS自旋;
2. 其它线程获取结果或取消后,重新唤醒CAS队列中等待的线程;
3. 再通过get()判断状态state;
4. 直至返回结果或(取消、超时、异常)为止。

三、FutureTask源码具体分析

1、FutureTask源码

通过定义整形状态值,判断state大小,这个思想很有意思,值得学习。

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    void run();
}
public class FutureTask<V> implements RunnableFuture<V> {

	// 最初始的状态是new 新建状态
	private volatile int state;
    private static final int NEW          = 0; // 新建状态
    private static final int COMPLETING   = 1; // 完成中
    private static final int NORMAL       = 2; // 正常执行完
    private static final int EXCEPTIONAL  = 3; // 异常
    private static final int CANCELLED    = 4; // 取消
    private static final int INTERRUPTING = 5; // 正在中断
    private static final int INTERRUPTED  = 6; // 已中断

	public V get() throws InterruptedException, ExecutionException {
	    int s = state;
	    // 任务还在执行中
	    if (s <= COMPLETING)
	        s = awaitDone(false, 0L);
	    return report(s);
	}
	
	private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
        	// 线程被中断,从等待队列中移除等待节点WaitNode,抛出中断异常
            if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }

            int s = state;
            // 任务已执行完毕或取消
            if (s > COMPLETING) {
            	// 如果已有等待节点WaitNode,将线程置空
                if (q != null)
                    q.thread = null;
                return s;
            }
            // 任务正在执行,让出时间片
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
            // 还未构造等待节点,则new一个新的等待节点
            else if (q == null)
                q = new WaitNode();
            // 未入队列,CAS尝试入队
            else if (!queued)
                queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                     q.next = waiters, q);
            // 如果有超时时间参数
            else if (timed) {
            	// 计算超时时间
                nanos = deadline - System.nanoTime();
                // 如果超时,则从等待队列中移除等待节点WaitNode,返回当前状态state
                if (nanos <= 0L) {
                    removeWaiter(q);
                    return state;
                }
                // 阻塞队列nanos毫秒
                LockSupport.parkNanos(this, nanos);
            }
            else
            	// 阻塞队列
                LockSupport.park(this);
        }
    }
    
	private V report(int s) throws ExecutionException {
		// 获取outcome中记录的返回结果
        Object x = outcome;
        // 如果执行完毕,返回结果
        if (s == NORMAL)
            return (V)x;
            // 如果大于等于取消状态,则抛出异常
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }
}
2、将异步方法的返回值改为Future<Integer>,将返回值放到new AsyncResult<>();中;
@Async("async-executor")
public void readXls(String filePath, String filename) {
    try {
    	// 此代码为简化关键性代码
        List<Future<Integer>> futureList = new ArrayList<>();
        for (int time = 0; time < times; time++) {
            Future<Integer> sumFuture = readExcelDataAsyncFutureService.readXlsCacheAsync();
            futureList.add(sumFuture);
        }
    }catch (Exception e){
        logger.error("readXlsCacheAsync---插入数据异常:",e);
    }
}
@Async("async-executor")
public Future<Integer> readXlsCacheAsync() {
    try {
        // 此代码为简化关键性代码
        return new AsyncResult<>(sum);
    }catch (Exception e){
        return new AsyncResult<>(0);
    }
}

3、通过Future<Integer>.get()获取返回值:

public static boolean getFutureResult(List<Future<Integer>> futureList, int excelRow) throws Exception{
    int[] futureSumArr = new int[futureList.size()];
    for (int i = 0;i<futureList.size();i++) {
        try {
            Future<Integer> future = futureList.get(i);
            while (true) {
                if (future.isDone() && !future.isCancelled()) {
                    Integer futureSum = future.get();
                    logger.info("获取Future返回值成功"+"----Future:" + future
                            + ",Result:" + futureSum);
                    futureSumArr[i] += futureSum;
                    break;
                } else {
                    logger.info("Future正在执行---获取Future返回值中---等待3秒");
                    Thread.sleep(3000);
                }
            }
        } catch (Exception e) {
            logger.error("获取Future返回值异常: ", e);
        }
    }
    
    boolean insertFlag = getInsertSum(futureSumArr, excelRow);
    logger.info("获取所有异步线程Future的返回值成功,Excel插入结果="+insertFlag);
    return insertFlag;
}

4、这里也可以通过新线程+Future获取Future返回值

不过感觉多此一举了,就当练习Future异步取返回值了~

public static Future<Boolean> getFutureResultThreadFuture(List<Future<Integer>> futureList, int excelRow) throws Exception {
    ExecutorService service = Executors.newSingleThreadExecutor();
    final boolean[] insertFlag = {false};
    service.execute(new Runnable() {
        public void run() {
            try {
                insertFlag[0] = getFutureResult(futureList, excelRow);
            } catch (Exception e) {
                logger.error("新线程+Future获取Future返回值异常: ", e);
                insertFlag[0] = false;
            }
        }
    });
    service.shutdown();
    return new AsyncResult<>(insertFlag[0]);
}

获取异步线程结果后,我们可以通过添加事务的方式,实现Excel入库操作的数据一致性。

但Future会造成主线程的阻塞,这个就很不友好了,有没有更优解呢?


在BUG中磨砺,在优化中成长

使用双异步后,从 191s 优化到 2s

增加索引 + 异步 + 不落地后,从 12h 优化到 15 min

使用懒加载 + 零拷贝后,程序的秒开率提升至99.99%

性能优化2.0,新增缓存后,程序的秒开率不升反降


🏆文章收录于:100天精通Java从入门到就业

全网最细Java零基础手把手入门教程,系列课程包括:Java基础、Java8新特性、Java集合、高并发、性能优化等,适合零基础和进阶提升的同学。

🏆哪吒多年工作总结:Java学习路线总结,搬砖工逆袭Java架构师

华为OD机试 2023B卷题库疯狂收录中,刷题点这里

刷的越多,抽中的概率越大,每一题都有详细的答题思路、详细的代码注释、样例测试,发现新题目,随时更新,全天CSDN在线答疑。文章来源地址https://www.toymoban.com/news/detail-817533.html

到了这里,关于使用双异步后,如何保证数据一致性?的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 如何保证缓存和数据库的数据一致性

    若数据库更新成功,删除缓存操作失败,则此后读到的都是缓存中过期的数据,造成不一致问题。 同删除缓存策略一样,若数据库更新成功缓存更新失败则会造成数据不一致问题。 若缓存更新成功数据库更新失败, 则此后读到的都是未持久化的数据。因为缓存中的数据是易

    2023年04月19日
    浏览(56)
  • MySQL和Redis如何保证数据一致性?

    由于缓存的高并发和高性能已经在各种项目中被广泛使用,在读取缓存这方面基本都是一致的,大概都是按照下图的流程进行操作: 但是在更新缓存方面,是更新完数据库再更新缓存还是直接删除缓存呢?又或者是先删除缓存再更新数据库?在这一点上就值得探讨了。 在实

    2024年02月01日
    浏览(51)
  • MySQL和Redis如何保证数据一致性

    MySQL与Redis都是常用的数据存储和缓存系统。为了提高应用程序的性能和可伸缩性,很多应用程序将MySQL和Redis一起使用,其中MySQL作为主要的持久存储,而Redis作为主要的缓存。在这种情况下,应用程序需要确保MySQL和Redis中的数据是同步的,以确保数据的一致性。 “数据一致

    2024年02月12日
    浏览(58)
  • 如何保证ES和数据库的数据一致性?

    在业务中,我们通常需要把数据库中的数据变更同步到ES中,那么如何保证数据库和ES的一致性呢?通常有以下几种做法: 双写 在代码中,对数据库和ES进行双写,并且先操作本地数据库,后操作ES,而且还需要把两个操作放到一个事务中:  在以上逻辑中,如果写数据库成功

    2024年04月28日
    浏览(49)
  • flink如何利用checkpoint保证数据状态一致性

    这本质上是一『尽力而为』的方法。保证数据或事件最多由应用程序中的所有算子处理一次。 这意味着如果数据在被流应用程序完全处理之前发生丢失,则不会进行其他重试或者重新发送。下图中的例子说明了这种情况。 应用程序中的所有算子都保证数据或事件至少被处理

    2024年02月21日
    浏览(48)
  • Redis---数据库和缓存如何保证一致性?

    用「读 + 写」请求的并发的场景来分析: 假如某个用户数据在缓存中不存在,请求 A 读取数据时从数据库中查询到年龄为 20,在未写入缓存中时另一个请求 B 更新数据。它更新数据库中的年龄为 21,并且清空缓存。这时请求 A 把从数据库中读到的年龄为 20 的数据写入到缓存

    2024年01月24日
    浏览(52)
  • Redis如何保证缓存和数据库一致性?

    现在我们在面向增删改查开发时,数据库数据量大时或者对响应要求较快,我们就需要用到Redis来拿取数据。 Redis:是一种高性能的内存数据库,它将数据以键值对的形式存储在内存中,具有读写速度快、支持多种数据类型、原子性操作、丰富的特性等优势。 优势: 性能极高

    2024年01月16日
    浏览(66)
  • mysql和redis如何保证数据库一致性

    如果对于小公司的单机服务器来说在更新和删除mysql数据的同时对redis缓存进行更新或者删除就行,一般有两个选择,例如: 先更新MySQL,后删除(或更新)Redis 先删除(或更新)Redis,后更新MySQL 但是不管使用其中哪种方式,都存在两个可能的问题: 由于第一步与第二步并不是原

    2023年04月24日
    浏览(46)
  • 并发情况如何实现加锁来保证数据一致性?

    ReentrantLock(可重入锁),指的是一个线程再次对已持有的锁保护的临界资源时,重入请求将会成功。 简单的与我们常用的Synchronized进行比较: ReentrantLock Synchronized 锁实现机制 依赖AQS 监视器模式 灵活性 支持响应超时、中断、尝试获取锁 不灵活 释放形式 必须显示调用unloc

    2024年02月05日
    浏览(70)
  • 如何保证数据库和缓存双写一致性?

    如何保证数据库和缓存双写一致性,是面试中经常被问的一个技术问题,程序汪推荐大家有必要好好研究一波 数据库和缓存(比如:redis)双写数据一致性问题,是一个跟开发语言无关的公共问题。尤其在高并发的场景下,这个问题变得更加严重。 我很负责的告诉大家,该问

    2024年01月18日
    浏览(57)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包