使用SynchronousQueue实现生产者/消费者

Java提供了许多用于并发支持的有用类中，有一个我想谈一谈： SynchronousQueue 。特别是，我想通过使用方便的SynchronousQueue作为交换机制来完成Producer / Consumer实现。

除非我们了解SynchronousQueue实现的内幕，否则可能不清楚为什么要使用这种类型的队列进行生产者/消费者通信。事实证明，这并不是我们过去通常考虑的队列。这个类比只是一个最多包含一个元素的集合。

为什么有用？好吧，有几个原因。从生产者的角度来看，只能将一个元素（或消息）存储到队列中。为了继续进行下一个元素（或消息），生产者应等到消费者使用队列中的当前元素。从使用者的角度来看，它只是轮询队列以查找下一个可用的元素（或消息）。很简单，但是最大的好处是：生产者发送消息的速度不能超过消费者处理消息的速度。

这是我最近遇到的用例之一：比较两个数据库表（可能只是巨大的），并检测其中包含不同数据或数据是否相同（副本）。 SynchronousQueue是解决此问题的便捷工具：它允许在自己的线程中处理每个表，并在从两个不同的数据库读取数据时补偿可能的超时/延迟。

让我们从定义比较功能开始，该功能接受源数据源和目标数据源以及表名（进行比较）。我正在使用Spring框架中非常有用的JdbcTemplate类，因为它非常好地抽象了处理连接和准备好的语句的所有无聊的细节。

public boolean compare( final DataSource source, final DataSource destination, final String table )  {final JdbcTemplate from = new  JdbcTemplate( source );final JdbcTemplate to = new JdbcTemplate( destination );
}

在进行任何实际数据比较之前，最好比较一下源数据库和目标数据库的表行数：

if( from.queryForLong('SELECT count(1) FROM ' + table ) != to.queryForLong('SELECT count(1) FROM ' + table ) ) {return false;
}

现在，至少知道表在两个数据库中包含相同数量的行，我们可以开始进行数据比较。该算法非常简单：

为源（生产者）和目标（消费者）数据库创建一个单独的线程
生产者线程从表中读取单行并将其放入SynchronousQueue
使用者线程还从表中读取单行，然后向队列询问要比较的可用行（必要时等待），最后比较两个结果集

使用另一大部分Java并发实用程序进行线程池，让我们定义一个具有固定线程数量的线程池（2）。

final ExecutorService executor = Executors.newFixedThreadPool( 2 );
final SynchronousQueue< List< ? > > resultSets = new SynchronousQueue< List< ? > >();

按照描述的算法，生产者功能可以表示为单个可调用项：

Callable< Void > producer = new Callable< Void >() {@Overridepublic Void call() throws Exception {from.query( 'SELECT * FROM ' + table,new RowCallbackHandler() {@Overridepublic void processRow(ResultSet rs) throws SQLException {try {                   List< ? > row = ...; // convert ResultSet to Listif( !resultSets.offer( row, 2, TimeUnit.MINUTES ) ) {throw new SQLException( 'Having more data but consumer has already completed' );}} catch( InterruptedException ex ) {throw new SQLException( 'Having more data but producer has been interrupted' );}}});return  null;}
};

由于Java语法，该代码有点冗长，但实际上并没有做很多事情。从表生成器读取的每个结果集都将转换为一个列表（由于是样板，因此省略了实现），并将其放入队列（ offer ）。如果队列不为空，则生产者将被阻止等待消费者完成工作。使用者可以分别表示为以下可调用对象：

Callable< Void > consumer = new Callable< Void >() {@Overridepublic Void call() throws Exception {to.query( 'SELECT * FROM ' + table,new RowCallbackHandler() {@Overridepublic void processRow(ResultSet rs) throws SQLException {try {List< ? > source = resultSets.poll( 2, TimeUnit.MINUTES );if( source == null ) {throw new SQLException( 'Having more data but producer has already completed' );}                                     List< ? > destination = ...; // convert ResultSet to Listif( !source.equals( destination ) ) {throw new SQLException( 'Row data is not the same' );}} catch ( InterruptedException ex ) {throw new SQLException( 'Having more data but consumer has been interrupted' );}}});return  null;}
};

使用者对队列执行反向操作：与其放入数据，不如将数据从队列中拉出（ poll ）。如果队列为空，则阻止消费者，等待生产者发布下一行。剩下的部分只是提交那些可调用对象以执行。 Future的get方法返回的任何异常都表明表不包含相同的数据（或者从数据库获取数据存在问题）：

List< Future< Void > > futures = executor.invokeAll( Arrays.asList( producer, consumer ) );for( final Future< Void > future: futures ) {future.get( 5, TimeUnit.MINUTES );}

参考： Andriy Redko {devmind}博客中的JCG合作伙伴 Andrey Redko 使用SynchronousQueue实现了生产者/消费者。