10 台のマシンを備えたシステムがあり、各マシンで特定のタスクを 1 つずつ同期順に実行する必要があります。基本的に、特定の時間にそのタスクを実行できるマシンは 1 台だけです。すでにConsul別の目的で使用していますが、これを使用できるConsulかどうかを考えていました。
詳細を読むと、各マシンがロックを取得し、作業を行ってからロックを解放しようとする領事でリーダー選挙を使用できるようです。作業が完了すると、ロックが解放され、他のマシンが再度ロックを取得して同じ作業を実行しようとします。このようにして、すべてが一度に 1 台のマシンで同期されます。
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()一度に1つずつ同期する必要があるということです。このブログと このブログに基づいて、ニーズに合わせて例を変更することにしました-
以下は私の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);に移動するとすぐに、自動的にメインメソッドに戻ります。Doing Some Work!プリントアウトに進むことはありません。CreateLock実際に動作しますか？私はそれがdata/lockconsulで作成され（そこにないため）、そのロックを取得しようとし、取得された場合は作業を行ってから他のマシンに解放することを期待していますか?
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);
}