[ Perl ] 多线程 并发编程

[ Perl ] 多线程 并发编程

https://www.cnblogs.com/yeungchie/

记录一些常用的 模块 / 方法 。

多线程

使用模块 threads

use 5.010;
use threads;

# 定义一个需要并发的子函数
sub func {
  my $id = shift;
  sleep 1;
  print "This is thread - $id
";
}

创建线程

  • new
sub start {
  my $id = shift;
  my $t = new threads &func, $id;
  return $t;
}
  • create
sub start {
  my $id = shift;
  my $t = create threads &func, $id;
  return $t;
}
  • async

可以不通过子函数来编写需要并发的过程,类似一个 “lambda” 。

sub start {
  my $id = shift;
  my $t = async { &func( $id ) };
  return $t;
}

线程收尸

  • 阻塞 join
&start( "join" )->join;
say "Done";

This is thread – join
Done
# 父线程被子线程阻塞,成功收尸。

  • 非阻塞 detach
&start( "detach" )->detach;
say "Done";

Done
# 由于非阻塞,父线程已经退出,子线程变成孤儿线程,无法收尸。

数据共享

使用模块 threads::shared

use threads::shared;

标记共享变量

有几种不同的写法

  • 依次标记 :shared
my $scalar :shared;
my @array  :shared;
my %hash   :shared;
  • 批量标记 :shared
my ( $scalar, @array, %hash ) :shared;
  • 用函数标记 share()
my ( $scalar, @array, %hash );
share $scalar;
share @array;
share %hash;

克隆 shared_clone

向共享的变量中加入新的元素时,需要注意的地方。

my @newArray = qw( YEUNG CHIE 1 2 3 );
my $clone = shared_clone [@newArray];
push @array, $clone;
$hash{ keyName } = $clone;

lock

多个线程同时编辑一个共享变量时,需要注意的地方。

经典的取钱问题:
1 – 输出额度 $amount = 500
2 – withdraw() 函数模拟取钱,每次取 300
3 – 当 $amount < 300 时,则无法取钱

  • 没加锁的情况
my $amount :shared = 500;

sub withdraw {
    unless ( $amount < 300 ) {
        sleep 1;  # 睡眠一秒模拟延迟
        $amount -= 300;
    }
}

# 这里两个线程模拟,两次取钱同时进行
my $t1 = new threads &withdraw;
my $t2 = new threads &withdraw;
$t1->join;
$t2->join;

say $amount;

-100
# 结果被取了两次,剩余额度为 -100

  • 加了锁的情况

调整一下子函数 withdraw(), 加个锁。

...
sub withdraw {
    lock $amount;
    unless ( $amount < 300 ) {
        sleep 1;
        $amount -= 300;
    }
}
...

200
# 结果正确

线程队列

使用模块 Thread::Queue

use Thread::Queue;

创建队列

my $queue = new Thread::Queue;

入队 enqueue

my $var = "YEUNG";
$queue->enqueue( $var );
$queue->enqueue( qw( CHIE 1 2 3 ) );

出队 dequeue

  • 默认出队一个项目
say $queue->dequeue;

YEUNG

  • 指定多个项目出队
say for $queue->dequeue( 3 );

CHIE
1
2

非阻塞出队 dequeue_nb

  • 如果是阻塞出队
my $queue = new Thread::Queue qw( YEUNG CHIE );
say while $_ = $queue->dequeue;

YEUNG
CHIE
# 程序会卡在这里,等待队列中新的项目加入

  • 使用非阻塞出队
my $queue = new Thread::Queue qw( YEUNG CHIE );
say while $_ = $queue->dequeue_nb;

YEUNG
CHIE

剩余 pending

pending 方法可以返回未出队的项目数量。

my $queue = new Thread::Queue qw( YEUNG CHIE );
say $queue->dequeue;
say $queue->pending;
say $queue->dequeue;
say $queue->pending;

YEUNG
1
CHIE
0

查看 peek

只是看看但是不出队。

my $queue = new Thread::Queue qw( YEUNG CHIE );
say $queue->peek;
say $queue->pending;
say $queue->peek( 2 );
say $queue->pending;

YEUNG
2
CHIE
2

入队结束 end

除了上面用 dequeue_nb 非阻塞出队,之外还可以用 end 方法来

my $queue = new Thread::Queue qw( YEUNG CHIE );
$queue->end;
say while $_ = $queue->dequeue;

# 这样虽然没有用 dequeue_nb 方法,程序也不会卡住了。

不过这个方法需要模块版本 >= 3.01,一般系统自带 Perl 是不支持的,但是我们也可以自己来实现这个效果:

  • 共享变量

    共享一个全局变量标记入队结束。

    my $endFlag :shared;
    
  • 生产者线程

    当入队结束时,$endFlag 赋值为真。

    $endFlag = 1;
    
  • 消费者线程

    循环操作非阻塞出队。

    while ( 1 ) {
        my $item = $queue->dequeue_nb;
        if ( defined $item ) {
            say $item;
        }
        else {
            # 当出队失败且入队结束时,退出循环
            last if $endFlag;
        }
    }
    

线程信号量

使用模块 Thread::Semaphore

use Thread::Semaphore;

线程池

使用模块 Thread::Pool

use Thread::Pool;

参考资料/拓展

  • threads – Perl interpreter-based threads – metacpan.org
  • threads::shared – Perl extension for sharing data structures between threads – metacpan.org
  • Thread::Queue – Thread-safe queues – metacpan.org
  • Thread::Semaphore – Thread-safe semaphores – metacpan.org
  • Thread::Pool – group of threads for performing similar jobs – metacpan.org
hmoban主题是根据ripro二开的主题,极致后台体验,无插件,集成会员系统
自学咖网 » [ Perl ] 多线程 并发编程