Java高并发之BlockingQueue

前言碎语

当系统流量负载比较高时，业务日志的写入操作也要纳入系统性能考量之内，如若处理不当，将影响系统的正常业务操作，之前写过一篇《spring boot通过MQ消费log4j2的日志》的博文，采用了RabbitMQ消息中间件来存储抗高并发下的日志，因为引入了中间件，操作使用起来可能没那么简便，今天分享使用多线程消费阻塞队列的方式来处理我们的海量日志

waht阻塞队列？

阻塞队列（BlockingQueue）是区别于普通队列多了两个附加操作的线程安全的队列。这两个附加的操作是：在队列为空时，获取元素的线程会等待队列变为非空。当队列满时，存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景，生产者是往队列里添加元素的线程，消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器，而消费者也只从容器里拿元素。

1.声明存储固定消息的队列

/*** Created by kl on 2017/3/20.* Content :销售操作日志队列*/
public class SalesLogQueue{//队列大小public static final int QUEUE_MAX_SIZE    = 1000;private static SalesLogQueue alarmMessageQueue = new SalesLogQueue();//阻塞队列private BlockingQueueblockingQueue = new LinkedBlockingQueue<>(QUEUE_MAX_SIZE);private SalesLogQueue(){}public static SalesLogQueue getInstance() {return alarmMessageQueue;}/*** 消息入队* @param salesLog* @return*/public boolean push(SalesLog salesLog) {return this.blockingQueue.add(salesLog);//队列满了就抛出异常，不阻塞}/*** 消息出队* @return*/public SalesLog poll() {SalesLog result = null;try {result = this.blockingQueue.take();} catch (InterruptedException e) {e.printStackTrace();}return result;}/*** 获取队列大小* @return*/public int size() {return this.blockingQueue.size();}
}

ps:因为业务原因，采用add的方式入队，队列满了就抛异常，不阻塞

2.消息入队

消息入队可以在任何需要保存日志的地方操作，如aop统一拦截日志处理，filter过滤请求日志处理，或者耦合的业务日志，记住，不阻塞入队操作，不然将影响正常的业务操作，如下为filter统一处理请求日志：

/*** Created by kl on 2017/3/20.* Content :访问请求拦截，保存操作日志*/
public class SalesLogFilter implements Filter {private RoleResourceService resourceService;@Overridepublic void init(FilterConfig filterConfig) throws ServletException {ServletContext context = filterConfig.getServletContext();ApplicationContext ctx = WebApplicationContextUtils.getWebApplicationContext(context);resourceService = ctx.getBean(RoleResourceService.class);}@Overridepublic void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {try {HttpServletRequest request = (HttpServletRequest) servletRequest;String requestUrl = request.getRequestURI();String requestType=request.getMethod();String ipAddress = HttpClientUtil.getIpAddr(request);Map resource=resourceService.getResource();String context=resource.get(requestUrl);//动态url正则匹配if(StringUtil.isNull(context)){for(Map.Entry entry:resource.entrySet()){String resourceUrl= entry.getKey();if(requestUrl.matches(resourceUrl)){context=entry.getValue();break;}}}SalesLog log=new SalesLog();log.setCreateDate(new Timestamp(System.currentTimeMillis()));log.setContext(context);log.setOperateUser(UserTokenUtil.currentUser.get().get("realname"));log.setRequestIp(ipAddress);log.setRequestUrl(requestUrl);log.setRequestType(requestType);SalesLogQueue.getInstance().push(log);}catch (Exception e){e.printStackTrace();}filterChain.doFilter(servletRequest, servletResponse);}@Overridepublic void destroy() {}
}

3.消息出队被消费

BlockingQueue是线程安全的，所以可以放心的在多个线程中去处理队列中的消息，如下代码声明了一个两个大小的固定线程池，并添加了两个线程去处理队列中的消息

/*** Created by kl on 2017/3/20.* Content :启动消费操作日志队列的线程*/
@Component
public class ConsumeSalesLogQueue {@AutowiredSalesLogService salesLogService;@PostConstructpublic void startrtThread() {ExecutorService e = Executors.newFixedThreadPool(2);//两个大小的固定线程池e.submit(new PollSalesLog(salesLogService));e.submit(new PollSalesLog(salesLogService));}class PollSalesLog implements Runnable {SalesLogService salesLogService;public PollSalesLog(SalesLogService salesLogService) {this.salesLogService = salesLogService;}@Overridepublic void run() {while (true) {try {SalesLog salesLog = SalesLogQueue.getInstance().poll();if(salesLog!=null){salesLogService.saveSalesLog(salesLog);}} catch (Exception e) {e.printStackTrace();}}}}
}

参考博文如下，对BlockingQueue队列更多了解，可读一读如下的博文：

http://blog.csdn.net/vernonzheng/article/details/8247564
http://www.infoq.com/cn/articles/java-blocking-queue
http://wsmajunfeng.iteye.com/blog/1629354

Java高并发之BlockingQueue相关推荐

Java高并发之设计模式，设计思想
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试文章作者:大道方圆 cnblogs.com/xdecode/p/913 ...
Java高并发之锁优化
本文主要讲并行优化的几种方式, 其结构如下: 锁优化减少锁的持有时间例如避免给整个方法加锁 1 public synchronized void syncMethod(){ 2 othercode ...
java高并发之魂：Synchronize
synchronize在高并发领域可谓是元老级别了,博主最近整理了一下synchronize的用法,和简单的概念. 概念对象锁:包括方法锁(默认锁对象为this当前实例对象)和同步代码块锁(自己指 ...
Java高并发之Hosee博客内容整理
Hosee博客博客高并发系列目录 [高并发Java 一] 前言 [高并发Java 二] 多线程基础 [高并发Java 三] Java内存模型和线程安全 [高并发Java 四] 无锁 [高并发Java ...
Java高并发之魂：synchronized深度解析
本文整理自慕课网的讲师悟空老师,教学地址:http://www.imooc.com/learn/1086 文章目录一.synchronized简介 1 synchronized作用 1.1 官方翻译 ...
Java高并发之CountDownLatch源码分析
概述 CountDownLatch 允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助.简单来说,就是 CountDownLatch 内部维护了一个计数器,每个线程完成自己的操作之后都 ...
java 高并发_Java 高并发之无锁(CAS)
Java 高并发之无锁(CAS) 本篇主要讲 Java中的无锁 CAS ,无锁顾名思义就是以不上锁的方式解决并发问题,而不使用synchronized 和 lock 等.. 1. Atomic 包 ...
Java高并发处理总结
自己参考大牛博客及视频写了一些关于并发的感悟,高并发的处理思路,无外乎以下几种 1 代码层面: 锁优化措施(见本文内容).尽量简化事务和减少事务 2 应用层面:缓存队列限流熔断 3数据库层面: ...
Java怎么避免重复订单_javaEE高并发之如何产生唯一不重复订单号
javaEE高并发之如何产生唯一不重复订单号 1.方案一:使用进程ID,线程ID,IP,MAC地址和时间戳进行拼接产生订单号 (1)如果没有并发,订单号只在一个线程内产生,那么由于程序是顺序执行的,不 ...

Java高并发之BlockingQueue

Java高并发之BlockingQueue相关推荐

最新文章

热门文章