使用Apache的Commons-Collection库中提供的BlockingBuffer实现数据异步处理

 

数据异步处理是常见的需求。假设一下场景： 

系统可能存在大量的数据并发写入某文件或某数据库表，而写入时还有相关业务处理，但是写入的结果对写入人来说毫不关心，无论是否写入成功都不重要，也不需要他立刻写入，慢慢写也可以。 

于是，我们可以用一种思路来解决，即创建一个队列，外部代码只负责将要处理的数据写入该队列，另有一个线程，尝试清空该队列，也就是处理数据。 

这就是一个比较简单的生产者和消费者的多线程模式，只是只有一个消费者，生产者很多。 


对于这种模式Apache开源库中的commons-collections中的Buffer，尤其是BlockingBuffer很好的实现了我们所说的这个队列。 

这个队列提供了一个"remove"方法，尝试获得队列中的数据，如果这个队列为空，这个方法将会阻塞，直到有数据为止。 

接下来，我们可以将这个逻辑封装为以下类：写道

public class DataDealer
{
    private final Buffer dataBuffer; // 数据buffer，因为使用匿名类，所以声明为final

    public DataDealer()
    {
        // 初始化阻塞队列，原始队列是一个无边界的先进先出队列，也是Apache的同一个库中提供的
        dataBuffer = BlockingBuffer.decorate( new UnboundedFifoBuffer(  ) );

        // 创建数据处理线程
        Thread dealThread = new Thread( new Runnable()
        {
            public void run()
            {
                while ( true )   // 永远尝试获得数据并处理，因此死循环
                {
                    System.out.println( "尝试获取要处理的数据" );
                    Object toDeal = dataBuffer.remove(); // 如果拿不到数据，就会阻塞
                    System.out.println( "dealing:" + toDeal ); // 真正的业务处理代码就在这里
                    System.out.println( "数据处理完毕" );
                }
            }
        }, "数据处理线程" );

        dealThread.setDaemon( true );  // 将这个数据处理线程设置为守护线程，如果主线程退出，那么个数据处理线程也自动退出
        dealThread.start();  // 启动这个线程
    }

    // 外部业务方法就简单了，调用这个方法提交一个要处理的对象，其实现逻辑只是将其加入队列而已
    public void addDataToWrite( Object toWhite )
    {
        dataBuffer.add( toWhite );
    }

}

 以上代码有一个注意事项，就是这个对象每创建一次就会发起一个线程，实际使用中，应该对其进一步改进，实现单例模式。