ruby并发编程

counter = 0

threads = 10.times.map do
  Thread.new do
    1000.times do
      counter += 1  # 非原子操作：读、加、写
    end
  end
end

threads.each(&:join)
puts counter  # 期望是 10000，但结果可能小于 10000

require 'concurrent-ruby'
class A
  # 引入Async module
  include Concurrent::Async

  def say_it(word)
    raise 'a is wrong' if word == 'a'

    sleep(1)
    puts "say #{word}"
    true
  end
end
# 异步方式调用
a = A.new.async.say_it('a') #异步执行
# value可传递timeout参数->超时秒数，执行到此时如果异步任务执行完，则返回结果，正在执行，则等待(阻塞)
puts a.value # 调用失败时 value为nil,如果是value!方法直接抛出异常
puts a.reason # error

b = A.new.async.say_it('b')
puts b.value # true
puts b.reason # 错误为nil

# 阻塞方式调用
c = A.new.await.say_it('c')
puts c

require 'concurrent-ruby'
task = Concurrent::ScheduledTask.new(2){ 'What does the fox say?' } # 2秒延时
puts task.class
puts task.state         #=> :unscheduled
task.execute # 开始执行
puts task.state         #=> pending

# wait for it...
sleep(3)

puts task.unscheduled? #=> false
puts task.pending?     #=> false
puts task.fulfilled?   #=> true
puts task.rejected?    #=> false
puts task.value        #=> 'What does the fox say?'

require 'concurrent-ruby'

# execution_interval 执行间隔（秒数），默认60秒
# interval_type 间隔方式默认fixed_delay
#               fixed_delay: 上一次执行结束和下一次执行开始之间的间隔
#               fixed_rate：上一次执行开始到下一次开始执行的间隔(如果执行时间超过间隔，则下一次执行将在前一次执行完成后立即开始。不会并发运行)
#                 
task = Concurrent::TimerTask.new(execution_interval: 1, interval_type: :fixed_rate){ t =  Time.now.sec; raise "aaa" if t % 5 == 0 ; puts t; t }

# 观察者类
class TaskObserver
  # time 执行时间
  # result 执行结果
  # ex 异常
  def update(time, result, ex)
    if result
      print "(#{time}) Execution successfully returned #{result}\n"
    else
      print "(#{time}) Execution failed with error #{ex}\n"
    end
  end
end

# 添加观察者
task.add_observer(TaskObserver.new) # 调用其update方法

task.execute # 开始执行

# 异步执行，防止主线程结束
gets
# 关闭
task.shutdown

require 'concurrent-ruby'

x =  Concurrent::Promises.
    future(2) { |v| raise 'aa' }.
    # 顺序执行
    then(&:succ).then{|x| x -= 1}.
    # 异常时
    rescue { |error| 999 }
puts x.result.inspect # 3



# 分支执行
head    = Concurrent::Promises.fulfilled_future(-1) # 
branch1 = head.then(&:abs).then(&:succ) # 分支1(绝对值 -> +1)
branch2 = head.then(&:succ).then(&:abs) # 分支2(+1 -> 绝对值)

puts head.value # -1
puts branch1.value # 2
puts branch2.value # 0

# 压缩分支
puts (branch1 & branch2).value.inspect # [2, 0]
# 任意分支
puts (branch1.then{raise 'a'} | branch2).value.inspect # 0

require 'concurrent-ruby'

pool = Concurrent::FixedThreadPool.new(5) # 最大5 threads

# 可以指定最大，最小线程数，回收空闲时间，
# pool = Concurrent::ThreadPoolExecutor.new(
#    min_threads: 5, # 最小线程数
#    max_threads: 5, # 最大线程数
#    idletime: 60, # 回收空闲时间
#    max_queue: 0 # 最大队列大小
#    fallback_policy: :abort # 等待队列满时策略：abort(异常)，discard(丢弃)，caller_runs(调用者线程执行)
# )
# 使用线程池执行异步任务
promises = (1..10).map do |i|
  # 在pool上执行
  Concurrent::Promises.future_on(pool, i) do |i|
    sleep 1 # 模拟耗时操作
    raise "x" if i %2 == 0
    puts "Task #{i} completed by #{Thread.current.object_id}"
    i * 2
  end
end
puts pool.running? # true
# 等待所有任务完成并获取结果
results = promises.map(&:value) # [2, nil, 6, nil, 10, nil, 14, nil, 18, nil]
puts pool.running? # true
puts "Results: #{results}"

# pool.wait_for_termination # 等待执行完关闭（但测试时报错，未知原因）
puts pool.running? # true
pool.shutdown # 需要手动关闭

require 'concurrent-edge'
# 输入channel，容量为2
in_c = Concurrent::Channel.new(capacity: 2)
# 输出channel
out_c = Concurrent::Channel.new(capacity: 2)

# 写入数据
Concurrent::Channel.go do
  10.times do |i|
    in_c.put(i)
  end
  in_c.close
end


Concurrent::Channel.go do
  loop do
    # 读取channel数据
    msg = ~ in_c
    break if msg.nil? # close时发送的nil数据

    out_c << (msg ** 2)
  end
  # 等效写法，each（忽略掉close的）
  # in_c.each do |msg|
  #  out_c << msg ** 2
  # end
  out_c.close # 关闭channel
end

# 读取
loop do
  v = ~out_c
  break if v.nil?
  puts v
end

NUM = 100000
THREAD_NUM = 4
BATCH_SIZE = NUM / THREAD_NUM 
def ractor_run(num)
  Ractor.new(num) do |start_index|
    sum = (start_index...start_index + BATCH_SIZE).inject(0) do |sum, i|
      sum += i ** 2
    end
    Ractor.yield(sum)
  end
end

def thread_run(num)
  Thread.new(num) do |start_index|
    (start_index...start_index+BATCH_SIZE).inject(0) do |sum, i|
      sum += i ** 2
    end
  end
end

def ractor_sum
  THREAD_NUM.times.map do |i|
    ractor_run(i * BATCH_SIZE)
  end.map(&:take).sum
end

def thread_sum
  THREAD_NUM.times.map do |i|
    thread_run(i * BATCH_SIZE)
  end.map{|t| t.join.value}.sum
end

def normal_sum
  (0...NUM).inject(0) do |sum, i|
    sum += i ** 2
  end
end

puts thread_sum

puts ractor_sum

puts normal_sum
require 'benchmark'

Benchmark.bmbm do |x|
  # sequential version
  x.report('normal'){ 100.times{normal_sum} }

  # parallel version with thread
  x.report('thread'){ 100.times{thread_sum}}
 
  # parallel version with ractors
  x.report('ractor'){ 100.times{ractor_sum} }
end

r1 = Ractor.new {Ractor.yield 'from 1';puts 'go on';4}
 r1.take # from 1
# print go on
r1.take # get 4

r1 = Ractor.new {msg = Ractor.receive;puts msg;5}
r1.send('ok') 
# print ok
r1.take # get 5

one = '3'.freeze
r = Ractor.new(one) do |one|
  one
end

two = r.take
puts one.object_id # 60
puts two.object_id # 60

one = []
r = Ractor.new(one) do |one|
  one
end

two = r.take
puts one.object_id # 80
puts two.object_id # 100

one = []
r = Ractor.new do
  x = Ractor.receive
  x
end
r.send(one, move: true)

two = r.take
puts two.object_id # 60
# 移动了，不能再使用
puts one.object_id #  `method_missing': can not send any methods to a moved object (Ractor::MovedError)