use DBI;
use threads;
use Thead::Queue;

my $Q = new Thread::Queue;

my $dbh = DBI->connect(...);
my $sth = $dbh->prepare("SELECT ...");
$sth->execute();


$thr1 = threads->new(\&some_sub, $Q );
$thr2 = threads->new(\&some_sub, $Q );

while( my $ref = $sth->fetchrow_hashref() ) {
   $Q->enqueue( join $;, %$ref );
   sleep 1 while $Q->pending > 10;
}

$Q->enqueue( (undef) x 2 );

$thr1->join; $thr2-join;

$sth->finish();
$dbh->disconnect();

sub some_sub {
    my ( $Q ) = @_;
    while( my %row = split $;, $Q->dequeue ) {
        #...
    } 
}