BT

如何利用碎片时间提升技术认知与能力? 点击获取答案

聊聊并发——生产者消费者模式

| 作者 方腾飞 关注 86 他的粉丝 发布于 2014年4月25日. 估计阅读时间: 23 分钟 | AICon 关注机器学习、计算机视觉、NLP、自动驾驶等20+AI热点技术和最新落地成功案例。

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里,生产者就是生产数据的线程,消费者就是消费数据的线程。在多线程开发当中,如果生产者处理速度很快,而消费者处理速度很慢,那么生产者就必须等待消费者处理完,才能继续生产数据。同样的道理,如果消费者的处理能力大于生产者,那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯,而通过阻塞队列来进行通讯,所以生产者生产完数据之后不用等待消费者处理,直接扔给阻塞队列,消费者不找生产者要数据,而是直接从阻塞队列里取,阻塞队列就相当于一个缓冲区,平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式,都会找一个第三者出来进行解耦,如工厂模式的第三者是工厂类,模板模式的第三者是模板类。在学习一些设计模式的过程中,如果先找到这个模式的第三者,能帮助我们快速熟悉一个设计模式。

生产者消费者模式实战

我和同事一起利用业余时间开发的Yuna工具中使用了生产者和消费者模式。首先我先介绍下Yuna工具,在阿里巴巴很多同事都喜欢通过邮件分享技术文章,因为通过邮件分享很方便,同学们在网上看到好的技术文章,复制粘贴发送就完成了分享,但是我们发现技术文章不能沉淀下来,对于新来的同学看不到以前分享的技术文章,大家也很难找到以前分享过的技术文章。为了解决这问题,我们开发了Yuna工具。Yuna取名自我喜欢的一款游戏最终幻想里的女主角。

首先我们申请了一个专门用来收集分享邮件的邮箱,比如share@alibaba.com,同学将分享的文章发送到这个邮箱,让同学们每次都抄送到这个邮箱肯定很麻烦,所以我们的做法是将这个邮箱地址放在部门邮件列表里,所以分享的同学只需要象以前一样向整个部门分享文章就行,Yuna工具通过读取邮件服务器里该邮箱的邮件,把所有分享的邮件下载下来,包括邮件的附件,图片,和邮件回复,我们可能会从这个邮箱里下载到一些非分享的文章,所以我们要求分享的邮件标题必须带有一个关键字,比如[内贸技术分享],下载完邮件之后,通过confluence的web service接口,把文章插入到confluence里,这样新同事就可以在confluence里看以前分享过的文章,并且Yuna工具还可以自动把文章进行分类和归档。

为了快速上线该功能,当时我们花了三天业余时间快速开发了Yuna1.0版本。在1.0版本中我并没有使用生产者消费模式,而是使用单线程来处理,因为当时只需要处理我们一个部门的邮件,所以单线程明显够用,整个过程是串行执行的。在一个线程里,程序先抽取全部的邮件,转化为文章对象,然后添加全部的文章,最后删除抽取过的邮件。代码如下:

public void extract() {
        logger.debug("开始" + getExtractorName() + "。。");
        //抽取邮件
        List<Article> articles = extractEmail();
        //添加文章
        for (Article article : articles) {
            addArticleOrComment(article);
        }
        //清空邮件
        cleanEmail();
        logger.debug("完成" + getExtractorName() + "。。");
    }

Yuna工具在推广后,越来越多的部门使用这个工具,处理的时间越来越慢,Yuna是每隔5分钟进行一次抽取的,而当邮件多的时候一次处理可能就花了几分钟,于是我在Yuna2.0版本里使用了生产者消费者模式来处理邮件,首先生产者线程按一定的规则去邮件系统里抽取邮件,然后存放在阻塞队列里,消费者从阻塞队列里取出文章后插入到conflunce里。代码如下:

public class QuickEmailToWikiExtractor extends AbstractExtractor {

private ThreadPoolExecutor      threadsPool;

private ArticleBlockingQueue<ExchangeEmailShallowDTO> emailQueue;

public QuickEmailToWikiExtractor() {
        emailQueue= new ArticleBlockingQueue<ExchangeEmailShallowDTO>();
        int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
        threadsPool = new ThreadPoolExecutor(corePoolSize, corePoolSize, 10l, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(2000));
    
    }

public void extract() {
        logger.debug("开始" + getExtractorName() + "。。");
        long start = System.currentTimeMillis();

        //抽取所有邮件放到队列里
        new ExtractEmailTask().start();

        // 把队列里的文章插入到Wiki
        insertToWiki();

        long end = System.currentTimeMillis();
        double cost = (end - start) / 1000;
        logger.debug("完成" + getExtractorName() + ",花费时间:" + cost + "秒");
    }

    

    /**
     * 把队列里的文章插入到Wiki
     */
    private void insertToWiki() {
        //登录wiki,每间隔一段时间需要登录一次
        confluenceService.login(RuleFactory.USER_NAME, RuleFactory.PASSWORD);

        while (true) {
            //2秒内取不到就退出
            ExchangeEmailShallowDTO email = emailQueue.poll(2, TimeUnit.SECONDS);
            if (email == null) {
                break;
            }
            threadsPool.submit(new insertToWikiTask(email));
        }
    }


     protected List<Article> extractEmail() {
        List<ExchangeEmailShallowDTO> allEmails = getEmailService().queryAllEmails();
        if (allEmails == null) {
            return null;
        }
        for (ExchangeEmailShallowDTO exchangeEmailShallowDTO : allEmails) {
            emailQueue.offer(exchangeEmailShallowDTO);
        }
        return null;
    }

    /**
     * 抽取邮件任务
     * 
     * @author tengfei.fangtf
     */
    public class ExtractEmailTask extends Thread {
        public void run() {
            extractEmail();
        }
    }
}

多生产者和多消费者场景

在多核时代,多线程并发处理速度比单线程处理速度更快,所以我们可以使用多个线程来生产数据,同样可以使用多个消费线程来消费数据。而更复杂的情况是,消费者消费的数据,有可能需要继续处理,于是消费者处理完数据之后,它又要作为生产者把数据放在新的队列里,交给其他消费者继续处理。如下图:

我们在一个长连接服务器中使用了这种模式,生产者1负责将所有客户端发送的消息存放在阻塞队列1里,消费者1从队列里读消息,然后通过消息ID进行hash得到N个队列中的一个,然后根据编号将消息存放在到不同的队列里,每个阻塞队列会分配一个线程来消费阻塞队列里的数据。如果消费者2无法消费消息,就将消息再抛回到阻塞队列1中,交给其他消费者处理。

以下是消息总队列的代码;

/**
 * 总消息队列管理
 * 
 * @author tengfei.fangtf
 */
public class MsgQueueManager implements IMsgQueue{

    private static final Logger              LOGGER             
 = LoggerFactory.getLogger(MsgQueueManager.class);


    /**
     * 消息总队列
     */
    public final BlockingQueue<Message> messageQueue;

    private MsgQueueManager() {
        messageQueue = new LinkedTransferQueue<Message>();
    }

    public void put(Message msg) {
        try {
            messageQueue.put(msg);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public Message take() {
        try {
            return messageQueue.take();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return null;
    }

}

启动一个消息分发线程。在这个线程里子队列自动去总队列里获取消息。

/**
     * 分发消息,负责把消息从大队列塞到小队列里
     * 
     * @author tengfei.fangtf
     */
    static class DispatchMessageTask implements Runnable {
        @Override
        public void run() {
            BlockingQueue<Message> subQueue;
            for (;;) {
                //如果没有数据,则阻塞在这里
                Message msg = MsgQueueFactory.getMessageQueue().take();
                //如果为空,则表示没有Session机器连接上来,
需要等待,直到有Session机器连接上来
                while ((subQueue = getInstance().getSubQueue()) == null) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
                //把消息放到小队列里
                try {
                    subQueue.put(msg);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

使用Hash算法获取一个子队列。

/**
     * 均衡获取一个子队列。
     * 
     * @return
     */
    public BlockingQueue<Message> getSubQueue() {
        int errorCount = 0;
        for (;;) {
            if (subMsgQueues.isEmpty()) {
                return null;
            }
            int index = (int) (System.nanoTime() % subMsgQueues.size());
            try {
                return subMsgQueues.get(index);
            } catch (Exception e) {
                //出现错误表示,在获取队列大小之后,队列进行了一次删除操作
                LOGGER.error("获取子队列出现错误", e);
                if ((++errorCount) < 3) {
                    continue;
                }
            }
        }
    }

使用的时候我们只需要往总队列里发消息。

//往消息队列里添加一条消息
        IMsgQueue messageQueue = MsgQueueFactory.getMessageQueue();
        Packet msg = Packet.createPacket(Packet64FrameType.
TYPE_DATA, "{}".getBytes(), (short) 1);
        messageQueue.put(msg);

小结

本章讲解了生产者消费者模式,并给出了实例。读者可以在平时的工作中思考下哪些场景可以使用生产者消费者模式,我相信这种场景应该非常之多,特别是需要处理任务时间比较长的场景,比如上传附件并处理,用户把文件上传到系统后,系统把文件丢到队列里,然后立刻返回告诉用户上传成功,最后消费者再去队列里取出文件处理。比如调用一个远程接口查询数据,如果远程服务接口查询时需要几十秒的时间,那么它可以提供一个申请查询的接口,这个接口把要申请查询任务放数据库中,然后该接口立刻返回。然后服务器端用线程轮询并获取申请任务进行处理,处理完之后发消息给调用方,让调用方再来调用另外一个接口拿数据。

另外Java中的线程池类其实就是一种生产者和消费者模式的实现方式,但是实现方法更高明。生产者把任务丢给线程池,线程池创建线程并处理任务,如果将要运行的任务数大于线程池的基本线程数就把任务扔到阻塞队列里,这种做法比只使用一个阻塞队列来实现生产者和消费者模式显然要高明很多,因为消费者能够处理直接就处理掉了,这样速度更快,而生产者先存,消费者再取这种方式显然慢一些。

我们的系统也可以使用线程池来实现多生产者消费者模式。比如创建N个不同规模的Java线程池来处理不同性质的任务,比如线程池1将数据读到内存之后,交给线程池2里的线程继续处理压缩数据。线程池1主要处理IO密集型任务,线程池2主要处理CPU密集型任务。


感谢张龙对本文的审校,感谢张龙对本文的策划。

给InfoQ中文站投稿或者参与内容翻译工作,请邮件至editors@cn.infoq.com。也欢迎大家通过新浪微博(@InfoQ)或者腾讯微博(@InfoQ)关注我们,并与我们的编辑和其他读者朋友交流。

评价本文

专业度
风格

您好,朋友!

您需要 注册一个InfoQ账号 或者 才能进行评论。在您完成注册后还需要进行一些设置。

获得来自InfoQ的更多体验。

告诉我们您的想法

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

不错 by 张章 鸥翔鱼游

看了此贴菜鸟又涨姿势了

更正 by Kwan Kent

生产者消费者模式解决的是两者速率不一致而产生的阻抗不匹配,解耦只不过是副产品而不是主要目的

Re: 更正 by 方 腾飞

是的。可能我没表达清楚,原文里也讲到是用来平衡了生产者和消费者的处理能力。

Re: 更正 by Kwan Kent

因为很多应聘者的答案都是解耦 所以我想有必要更正一下

ab by char kaka

对于即时性要求比较强的并发,有什么好的处理方案?

入门贴? by xie loufer

这类文章应该发到CSDN,而不是InfoQ

挺通俗的解决方案,不错,实在 by 曦 杨

挺通俗的解决方案,不错,实在

挺通俗的解决方案,不错,实在 by 曦 杨

挺通俗的解决方案,不错,实在

问题 by Noah Li

"生产者1负责将所有客户端发送的消息存放在阻塞队列1里,消费者1从队列里读消息,然后通过消息ID进行hash得到N个队列中的一个,然后根据编号将消息存放在到不同的队列里,每个阻塞队列会分配一个线程来消费阻塞队列里的数据。"
不大明白,这里为什么不直接HASH到各个子队列?

Re: 问题 by 方 腾飞

主队列主要负责数据处理的统一调度。各子队列需要保持做事的单纯性,就是拿数据并处理数据。某个子队列如果处理不过来,只需要把消息重新发回给主队列,自己不需要计算应该发给哪个子队列更合适。如果增加了新的子队列,也由主队列来平衡各个子队列的处理效率。

这种文章也能发到infoQ? by du yaming

也太入门级别了吧

Re: 问题 by 李 鹏

Re: 这种文章也能发到infoQ? by 李 鹏

介绍的还是不错的,算是入门级别,发在哪儿不重要,关键是说得清楚就行

作为一只菜鸟我想说 by Leon Wood

作为一只菜鸟我想说是不是用node.js做会更好?或者这种获取->反馈->再处理就是在实现一种类似用异步回调?不知道说的对不对.....

关于线程池的使用 by lee noah

我想请问作者,您文章中最后提到:
。比如创建N个不同规模的Java线程池来处理不同性质的任务,
我比较困惑的是,在一个java进程里面,创建一个线程池处理所有任务好呢,还是像您说的这种方案好。在我的理解,线程池就是一个执行处理、运算的公共资源,创建多个线程池的好处在哪儿,仅仅是在程序易读性上好吗?

Re: 关于线程池的使用 by 方 腾飞

看情况。一般情况下使用一个线程池就可以。如果任务比较多,性质也不一样,又不能互相干扰,可以考虑使用多个线程池,比如tomcat对于所有的请求,要做两件事情,第一个建立请求,第二个处理请求,而建立请求必须很快,所以用了一个单独的线程池来做,做完之后丢给另外一个线程池,另外一个线程池里的线程负责处理请求,可能会处理得很慢,多个线程池主要是避免两个不同性质的任务互相干扰。

Re: ab by 方 腾飞

并发度小的话,可以建立长连接。

Re: ab by 方 腾飞

并发度小的话,可以建立长连接。

Re: 入门贴? by 吴 杰峰

为什么要发到CSDN,发到那边真正的好文就石城大海了!

Re: ab by Zhang Boyce

同问!

Good!但是遇到这种并发处理的情景,有一点我都很疑惑: by hu dk

如果中途处理失败了,怎么恢复处理呢?是从头开始处理,还是记录一个检查点,下次启动时,从检查点开始执行?

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

允许的HTML标签: a,b,br,blockquote,i,li,pre,u,ul,p

当有人回复此评论时请E-mail通知我

21 讨论

登陆InfoQ,与你最关心的话题互动。


找回密码....

Follow

关注你最喜爱的话题和作者

快速浏览网站内你所感兴趣话题的精选内容。

Like

内容自由定制

选择想要阅读的主题和喜爱的作者定制自己的新闻源。

Notifications

获取更新

设置通知机制以获取内容更新对您而言是否重要

BT