如何使用 Consul 一次在一台机器上执行同步任务?

用户1950349

我有一个有 10 台机器的系统,我需要在每台机器上按同步顺序一个接一个地执行某个任务。基本上只有一台机器应该在特定时间完成这项任务。我们已经Consul用于其他目的,但我在想我们也可以Consul用来做这个吗?

我阅读了更多关于它的信息,看起来我们可以使用带有 consul 的领导者选举,其中每台机器都会尝试获取锁,完成工作然后释放锁。一旦工作完成,它将释放锁,然后其他机器将再次尝试获取锁并执行相同的工作。这样,所有内容都将一次在一台机器上同步。

我决定使用这个已经内置了这个功能的C# PlayFab ConsulDotNet ,但如果有更好的选择,我也愿意接受。Action我的代码库中的以下方法几乎通过观察者机制在每台机器上同时调用。

 private void Action() {
    // Try to acquire lock using Consul.
    // If lock acquired then DoTheWork() otherwise keep waiting for it until lock is acquired.
    // Once work is done, release the lock
    // so that some other machine can acquire the lock and do the same work.
 }

现在在上述方法中,我需要做以下事情-

  • 尝试获取锁。如果您无法获得锁,请等待它,因为其他机器可能在您之前抢到了它。
  • 如果获得锁,则 DoTheWork()。
  • 工作完成后,释放锁,以便其他机器可以获取锁并执行相同的工作。

想法是所有 10 台机器都应该DoTheWork()以同步顺序一次一基于此博客和此博客,我决定修改他们的示例以满足我们的需求-

下面是我的LeaderElectionService课:

public class LeaderElectionService
{
    public LeaderElectionService(string leadershipLockKey)
    {
        this.key = leadershipLockKey;
    }

    public event EventHandler<LeaderChangedEventArgs> LeaderChanged;
    string key;
    CancellationTokenSource cts = new CancellationTokenSource();
    Timer timer;
    bool lastIsHeld = false;
    IDistributedLock distributedLock;

    public void Start()
    {
        timer = new Timer(async (object state) => await TryAcquireLock((CancellationToken)state), cts.Token, 0, Timeout.Infinite);
    }

    private async Task TryAcquireLock(CancellationToken token)
    {
        if (token.IsCancellationRequested)
            return;
        try
        {
            if (distributedLock == null)
            {
                var clientConfig = new ConsulClientConfiguration { Address = new Uri("http://consul.host.domain.com") };
                ConsulClient client = new ConsulClient(clientConfig);
                distributedLock = await client.AcquireLock(new LockOptions(key) { LockTryOnce = true, LockWaitTime = TimeSpan.FromSeconds(3) }, token).ConfigureAwait(false);
            }
            else
            {
                if (!distributedLock.IsHeld)
                {
                    await distributedLock.Acquire(token).ConfigureAwait(false);
                }
            }
        }
        catch (LockMaxAttemptsReachedException ex)
        {
            //this is expected if it couldn't acquire the lock within the first attempt.
            Console.WriteLine(ex.Stacktrace);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Stacktrace);
        }
        finally
        {
            bool lockHeld = distributedLock?.IsHeld == true;
            HandleLockStatusChange(lockHeld);
            //Retrigger the timer after a 10 seconds delay (in this example). Delay for 7s if not held as the AcquireLock call will block for ~3s in every failed attempt.
            timer.Change(lockHeld ? 10000 : 7000, Timeout.Infinite);
        }
    }

    protected virtual void HandleLockStatusChange(bool isHeldNew)
    {
        // Is this the right way to check and do the work here?
        // In general I want to call method "DoTheWork" in "Action" method itself
        // And then release and destroy the session once work is done.
        if (isHeldNew)
        {
            // DoTheWork();
            Console.WriteLine("Hello");
            // And then were should I release the lock so that other machine can try to grab it?
            // distributedLock.Release();
            // distributedLock.Destroy();
        }

        if (lastIsHeld == isHeldNew)
            return;
        else
        {
            lastIsHeld = isHeldNew;
        }

        if (LeaderChanged != null)
        {
            LeaderChangedEventArgs args = new LeaderChangedEventArgs(lastIsHeld);
            foreach (EventHandler<LeaderChangedEventArgs> handler in LeaderChanged.GetInvocationList())
            {
                try
                {
                    handler(this, args);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Stacktrace);
                }
            }
        }
    }
}

下面是我的LeaderChangedEventArgs课:

public class LeaderChangedEventArgs : EventArgs
{
    private bool isLeader;

    public LeaderChangedEventArgs(bool isHeld)
    {
        isLeader = isHeld;
    }

    public bool IsLeader { get { return isLeader; } }
}

在上面的代码中,我的用例可能不需要很多部分,但想法是相同的。

问题陈述

现在在我的Action方法中,我想使用上面的类并在获取锁后立即执行任务,否则继续等待锁。工作完成后,释放并销毁会话,以便其他机器可以抓取它并完成工作。我有点困惑如何在下面的方法中正确使用上面的类。

 private void Action() {
    LeaderElectionService electionService = new LeaderElectionService("data/process");
    // electionService.LeaderChanged += (source, arguments) => Console.WriteLine(arguments.IsLeader ? "Leader" : "Slave");
    electionService.Start();

    // now how do I wait for the lock to be acquired here indefinitely
    // And once lock is acquired, do the work and then release and destroy the session
    // so that other machine can grab the lock and do the work
 }

我最近开始工作,C#所以这就是为什么有点困惑如何通过使用Consul和这个库在生产中有效地完成这项工作

更新

我按照你的建议尝试了下面的代码,我想我之前也尝试过这个,但是出于某种原因,一旦它转到这一行 await distributedLock.Acquire(cancellationToken);,它就会自动返回到 main 方法。它永远不会前进到我的Doing Some Work!打印输出。是否CreateLock实际工作?我期待它会data/lock在 consul 上创建(因为它不存在),然后尝试获取它的锁,如果获取了,然后做工作然后将它释放给其他机器?

private static CancellationTokenSource cts = new CancellationTokenSource();

public static void Main(string[] args)
{
    Action(cts.Token);
    Console.WriteLine("Hello World");
}

private static async Task Action(CancellationToken cancellationToken)
{
    const string keyName = "data/lock";

    var clientConfig = new ConsulClientConfiguration { Address = new Uri("http://consul.test.host.com") };
    ConsulClient client = new ConsulClient(clientConfig);
    var distributedLock = client.CreateLock(keyName);

    while (true)
    {
        try
        {
            // Try to acquire lock
            // As soon as it comes to this line,
            // it just goes back to main method automatically. not sure why
            await distributedLock.Acquire(cancellationToken);

            // Lock is acquired
            // DoTheWork();
            Console.WriteLine("Doing Some Work!");

            // Work is done. Jump out of loop to release the lock
            break;
        }
        catch (LockHeldException)
        {
            // Cannot acquire the lock. Wait a while then retry
            await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
        }
        catch (Exception)
        {
            // TODO: Handle exception thrown by DoTheWork method

            // Here we jump out of the loop to release the lock
            // But you can try to acquire the lock again based on your requirements
            break;
        }
    }

    // Release and destroy the lock
    // So that other machine can grab the lock and do the work
    await distributedLock.Release(cancellationToken);
    await distributedLock.Destroy(cancellationToken);
}
Han Zhao

海事组织,LeaderElectionService从这些博客是你的情况是一种矫枉过正。

更新 1

没有必要做while循环,因为:

  1. ConsulClient 是局部变量
    • 无需检查IsHeld财产
  2. Acquire 将无限期阻止,除非
    • 设置为LockTryOnceLockOptions
    • 将超时设置为 CancellationToken

旁注,在调用分布式锁(参考Destroy没有必要调用方法Release

private async Task Action(CancellationToken cancellationToken)
{
    const string keyName = "YOUR_KEY";

    var client = new ConsulClient();
    var distributedLock = client.CreateLock(keyName);

    try
    {
        // Try to acquire lock
        // NOTE:
        //   Acquire method will block indefinitely unless
        //     1. Set LockTryOnce = true in LockOptions
        //     2. Pass a timeout to cancellation token
        await distributedLock.Acquire(cancellationToken);

        // Lock is acquired
        DoTheWork();
    }
    catch (Exception)
    {
        // TODO: Handle exception thrown by DoTheWork method
    }

    // Release the lock (not necessary to invoke Destroy method), 
    // so that other machine can grab the lock and do the work
    await distributedLock.Release(cancellationToken);
}

更新 2

OP 的代码只返回Main方法的原因是,Action方法不是 awaited如果您使用 C# 7.1,则可以使用async Main,并await使用Action方法。

public static async Task Main(string[] args)
{
    await Action(cts.Token);
    Console.WriteLine("Hello World");
}

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章

如何在 VMware 和 Virtualbox 上运行同一台机器?

在一台机器上使用sixxs子网

如何在一台机器上编译幻象,然后在另一台机器上“移植”?

在一台机器上运行多个jmeter任务,如何指定一个端口来停止jmeter测试

如何配置Apache在同一台机器上与FE和BE一起使用?

一台Linux机器上的分段错误,但使用C ++代码不在另一台机器上

如何在一台机器上使用socat测试多播

如何在一台机器上使用GitLab和Gitlab-Mattermost?

如何从一台机器到另一台机器执行批量插入到 SQLServer 中?

如何将网站从一台机器上的 IIS 7.5 迁移到另一台机器上的 IIS 8.5?

如何在一台机器上运行一个外壳的多个版本

如何修复.bashrc在配置与另一台机器相同的一台机器上没有颜色的问题?

如何修复仅一台机器损坏的SSH?

如何在同一台机器上运行多个 Apache Ignite 节点?

如何在同一台机器上设置不同的 Scala 版本?

如何在同一台机器上安装opencv 2.4和opencv 3

如何从另一台机器查看托管在 ubuntu 上的 Web 应用程序?

如何配置Akka Pub / Sub在同一台机器上运行?

如何从本地网络上的另一台机器访问Docker容器

如何在同一台机器上运行多个硒测试套件

如何在Ubuntu的另一台机器上移动Alpine邮件?

如何在一台机器上设置多个gemfire / geode WAN集群进行测试?

如何与群管理器在同一台机器上运行docker群节点?

如何在另一台机器上克隆RStudio环境

如何在一台机器上设置多个Web客户端

如何收集要在另一台机器上提交的错误数据?

同一台机器上的两个Tomcat实例使用相同的日志文件

可以在一台机器上使用GitHub和GitLab吗?

在另一台机器上使用gulp项目