基于redis实现的锁

代码：

class MsgSchedule
  attr_reader :lock_key, :expire_in, :redis

  def initialize(lock_key, opts = {})
    raise 'lock_key must be exists!' if lock_key.blank?

    @lock_key  = lock_key
    @expire_in = opts.delete(:expire_in) || 1.minute
    @redis     = opts.delete(:redis) || MSG_BUFFERS_REDIS
    @opts      = opts
  end

  def synchronize
    set_owner_flag
    return unless lock

    begin
      yield self
    ensure
      unlock
    end

    true
  end

  private

  def lock
    @locked = @redis.set(lock_key, owner_flag, nx: true, ex: expire_in)
    @locked ||= (unlock_by_expired && lock)
  end

  def unlock
    unlock_by_myself || unlock_by_expired
  end

  def unlock_by_myself
    val = @redis.get(lock_key)
    val == owner_flag && !!@redis.del(lock_key)
  end

  def unlock_by_expired
    val = @redis.get(lock_key)
    val.nil? || (val.to_f + expire_in <= Time.current.to_f) && !!@redis.del(lock_key)
  end

  def owner_flag
    @owner_flag
  end

  def set_owner_flag
    @owner_flag = Time.current.to_f.to_s
  end
end

 

测试代码：

@a, @b, @k, @ex = (1..50).to_a, [], 'test/ms', 1.seconds

5.times.map {
  Thread.new {
    ms  = MsgSchedule.new(@k, expire_in: @ex)
    @ss = false

    loop {
      puts "数据 ====> #{@a.size} == #{@b.size}"
      break if @a.uniq.size <= 1
      puts "等待加锁 ==> #{sleep(0.1)}" if @ss

      ms.synchronize {
        x = @a.find { |x| x > -1 }
        sleep(rand(3) / 10.0)
        @ss = x.nil?
        unless x.nil?
          @b << x
          @a[@a.index(x)] = -1
        end
      }
    }
  }
}.join

puts @a, @b