建设厅网站用户名和密码,传媒公司网站建设策划,咨询机构,一个公司做两个网站可以吗在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里#xff0c;生产者就是生产数据的线程#xff0c;消费者就是消费数据的线程。在…在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。
为什么要使用生产者和消费者模式
在线程世界里生产者就是生产数据的线程消费者就是消费数据的线程。在多线程开发当中如果生产者处理速度很快而消费者处理速度很慢那么生产者就必须等待消费者处理完才能继续生产数据。同样的道理如果消费者的处理能力大于生产者那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。
什么是生产者消费者模式
生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯而通过阻塞队列来进行通讯所以生产者生产完数据之后不用等待消费者处理直接扔给阻塞队列消费者不找生产者要数据而是直接从阻塞队列里取阻塞队列就相当于一个缓冲区平衡了生产者和消费者的处理能力。 这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式都会找一个第三者出来进行解耦如工厂模式的第三者是工厂类模板模式的第三者是模板类。在学习一些设计模式的过程中如果先找到这个模式的第三者能帮助我们快速熟悉一个设计模式。
生产者消费者模式实战
我和同事一起利用业余时间开发的Yuna工具中使用了生产者和消费者模式。首先我先介绍下Yuna工具在阿里巴巴很多同事都喜欢通过邮件分享技术文章因为通过邮件分享很方便同学们在网上看到好的技术文章复制粘贴发送就完成了分享但是我们发现技术文章不能沉淀下来对于新来的同学看不到以前分享的技术文章大家也很难找到以前分享过的技术文章。为了解决这问题我们开发了Yuna工具。Yuna取名自我喜欢的一款游戏最终幻想里的女主角。
首先我们申请了一个专门用来收集分享邮件的邮箱比如sharealibaba.com同学将分享的文章发送到这个邮箱让同学们每次都抄送到这个邮箱肯定很麻烦所以我们的做法是将这个邮箱地址放在部门邮件列表里所以分享的同学只需要象以前一样向整个部门分享文章就行Yuna工具通过读取邮件服务器里该邮箱的邮件把所有分享的邮件下载下来包括邮件的附件图片和邮件回复我们可能会从这个邮箱里下载到一些非分享的文章所以我们要求分享的邮件标题必须带有一个关键字比如内贸技术分享下载完邮件之后通过confluence的web service接口把文章插入到confluence里这样新同事就可以在confluence里看以前分享过的文章并且Yuna工具还可以自动把文章进行分类和归档。
为了快速上线该功能当时我们花了三天业余时间快速开发了Yuna1.0版本。在1.0版本中我并没有使用生产者消费模式而是使用单线程来处理因为当时只需要处理我们一个部门的邮件所以单线程明显够用整个过程是串行执行的。在一个线程里程序先抽取全部的邮件转化为文章对象然后添加全部的文章最后删除抽取过的邮件。代码如下
public void extract() {logger.debug(开始 getExtractorName() 。。);//抽取邮件ListArticle 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 ArticleBlockingQueueExchangeEmailShallowDTO emailQueue;public QuickEmailToWikiExtractor() {emailQueue new ArticleBlockingQueueExchangeEmailShallowDTO();int corePoolSize Runtime.getRuntime().availableProcessors() * 2;threadsPool new ThreadPoolExecutor(corePoolSize, corePoolSize, 10l, TimeUnit.SECONDS,new LinkedBlockingQueueRunnable(2000));}public void extract() {logger.debug(开始 getExtractorName() 。。);long start System.currentTimeMillis();//抽取所有邮件放到队列里new ExtractEmailTask().start();// 把队列里的文章插入到WikiinsertToWiki();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 ListArticle extractEmail() {ListExchangeEmailShallowDTO 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 BlockingQueueMessage messageQueue;private MsgQueueManager() {messageQueue new LinkedTransferQueueMessage();}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 {Overridepublic void run() {BlockingQueueMessage 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 BlockingQueueMessage 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密集型任务。
