ZooKeeper Distributed lock

雨点打透心脏的1/2处 2023-08-17 17:46 109阅读 0赞

*  https://segmentfault.com/a/1190000016351095
 *  http://www.dengshenyu.com/java/%E5%88%86%E5%B8%83%E5%BC%8F%E7%B3%BB%E7%BB%9F/2017/10/23/zookeeper-distributed-lock.html

### Test ###

Enumerable.Range(1, 5).ToList().ForEach(i =>
                 {
                     Task.Run(() =>
                     {
    
                         var lockHelper = new ZooKeeperLockHelper("localhost:5181");
                         lockHelper.OnAcquireLock += (id) =>
                         {
                             var random = new Random().Next(10);
                             Log.Debug("NodeId {@id} executing.....Sleep {@ms} ms", id, random * 1000);
                             
                             Thread.Sleep(random * 1000);
                             Log.Debug("NodeId {@id} executing success", id);
    
                             return Task.CompletedTask;
                         };
    
                         lockHelper.AcquireLock();
                        
                    });
    
                 });

using org.apache.zookeeper;
    using Serilog;
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Text.RegularExpressions;
    using System.Threading;
    using System.Threading.Tasks;
    
    namespace RedisDemo
    {
        public class ZooKeeperLockHelper : Watcher, IDisposable
        {
    
            #region event
    
            public event Func<long, Task> OnAcquireLock;
            #endregion
    
            private bool _disposed;
    
            private ZooKeeper _zooKeeper;
            private Event.KeeperState _currentState;
            private AutoResetEvent _notifyEvent = new AutoResetEvent(false);
    
    
            private string _connectionString;
    
            private bool _hasAcquireLock;
            private string _lockPath;
            private long _currentNodeId;
    
            private static readonly string DEFAULT_PATH = "/zk";
            private static readonly string NODE_NAME = "node-";
    
            public ZooKeeperLockHelper(string connectionString)
            {
                _connectionString = connectionString;
                this.Initialize(_connectionString, TimeSpan.FromSeconds(60));
            }
    
            public void AcquireLock(string path = "")
            {
                if (this._hasAcquireLock)
                {
                    FireAcquireLock(this._currentNodeId).Wait();
                    return;
                }
    
                if (!WaitConnected(TimeSpan.FromSeconds(10)))
                {
                    throw new Exception($"{_connectionString} Cannot Connect ZooKeeper");
                }
    
                _lockPath = path;
                if (string.IsNullOrEmpty(_lockPath))
                {
                    _lockPath = DEFAULT_PATH;
                }
    
                var nodePath = _lockPath + "/" + NODE_NAME;
    
                var spath = this._zooKeeper.createAsync(
                    nodePath, Encoding.UTF8.GetBytes("data"),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL).Result;
                this._currentNodeId = ParseNodeId(spath);
    
                var reuslt = this._zooKeeper.getChildrenAsync(_lockPath, true).GetAwaiter().GetResult();
                Log.Debug("#-> Begin Acquire Lock CurrentId {@id}", _currentNodeId);
    
                if (this.IsMinNodeId(reuslt, this._currentNodeId))
                {
                    lock (this)
                    {
                        if (!this._hasAcquireLock)
                        {
                            Log.Debug("NodeId {@id} Direct Acquire Lock", _currentNodeId);
                            this._hasAcquireLock = true;
                            this.FireAcquireLock(this._currentNodeId).Wait();
                        }
    
                    }
                }
    
            }
    
            protected bool IsMinNodeId(ChildrenResult childrenResult, long nodeId)
            {
                if (nodeId == 0 || childrenResult == null || childrenResult.Children.Count == 0)
                    return false;
    
                var nodeIds = new List<long>();
    
                foreach (var item in childrenResult.Children)
                {
                    nodeIds.Add(ParseNodeId(item));
                }
    
                if (nodeIds.Count > 0 && nodeIds.Min() == nodeId)
                {
                    return true;
    
                }
                return false;
            }
    
            protected long ParseNodeId(string path)
            {
                var m = Regex.Match(path, "(\\d+)");
                if (m.Success)
                {
                    return long.Parse(m.Groups[0].Value);
                }
                return 0L;
            }
    
            protected void Initialize(String connectionString, TimeSpan sessionTimeout)
            {
                this._zooKeeper = new ZooKeeper(connectionString, (int)sessionTimeout.TotalMilliseconds, this);
            }
    
            public Task FireAcquireLock(long id)
            {
                this.OnAcquireLock(id).Wait();
                this.CloseConnection();
                Log.Debug("NodeId {@id} Close ZooKeeper Success", id);
                return Task.CompletedTask;
            }
    
            public bool WaitConnected(TimeSpan timeout)
            {
                var continueWait = false;
                while (this._currentState != Event.KeeperState.SyncConnected)
                {
                    continueWait = _notifyEvent.WaitOne(timeout);
                    if (!continueWait)
                    {
                        return false;
                    }
                }
                return true;
            }
    
            protected void CloseConnection()
            {
                if (_disposed)
                {
                    return;
                }
                _disposed = true;
    
                if (_zooKeeper != null)
                {
                    try
                    {
                        this._zooKeeper.closeAsync().ConfigureAwait(false).GetAwaiter().GetResult();
                    }
                    catch { }
    
                }
            }
    
            #region Watcher Impl
    
            public override Task process(WatchedEvent @event)
            {
                if (@event.getState() == Event.KeeperState.SyncConnected)
                {
                    if (String.IsNullOrEmpty(@event.getPath()))
                    {
                        this._currentState = @event.getState();
                        this._notifyEvent.Set();
                    }
    
                    var path = @event.getPath();
                    if (!string.IsNullOrEmpty(path))
                    {
                        if (path.Equals(this._lockPath))
                        {
                            Log.Debug("NodeId {@id} Start Watcher Callback", this._currentNodeId);
    
                            if (this._hasAcquireLock)
                            {
                                Log.Debug("NodeId {@id} Has Acquire Lock return", this._currentNodeId);
                                return Task.CompletedTask;
                            }
    
                            Task.Run(() =>
                            {
                                var childrenResult = _zooKeeper.getChildrenAsync(this._lockPath, this).Result;
    
                                if (IsMinNodeId(childrenResult, this._currentNodeId))
                                {
                                    lock (this)
                                    {
                                        if (!this._hasAcquireLock)
                                        {
                                            Log.Debug("NodeId {@id} Acquire Lock", this._currentNodeId);
                                            this._hasAcquireLock = true;
                                            this.FireAcquireLock(this._currentNodeId).Wait();
                                        }
                                    }
                                }
                            });
    
                            //_zooKeeper.getChildrenAsync(_lockPath, this);
    
                        }
                    }
                }
    
                return Task.CompletedTask;
            }
    
    
    
            public void Dispose()
            {
                this.CloseConnection();
            }
    
            #endregion
        }
    }

转载于:https://www.cnblogs.com/byxxw/p/11352564.html