curator分散シーケンス番号ジェネレータ
2937 ワード
1、zookeeperでロックディレクトリを作成することを前提とする
2、次のコードはcuratorが提供した分散ロックを使用して、次のシーケンス番号の自己増加を試みます.ここではデータの上界を考慮していません.ただの例です.
2、次のコードはcuratorが提供した分散ロックを使用して、次のシーケンス番号の自己増加を試みます.ここではデータの上界を考慮していません.ただの例です.
package mytest;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
/**
* distributed sequential number generator based on zookeeper
* prediction: zookeeper has the lock path already. this program just watch that path
* @author sing
*
*/
public class DistrSeqGen {
private CuratorFramework curator;
private InterProcessMutex mutex ;
private String lockPath;
private DistrSeqGen(String serverLoc,String lockPath) {
// synchronized(DistrSeqGen.class) {
curator = CuratorFrameworkFactory.builder()
.connectString(serverLoc)
.retryPolicy(new ExponentialBackoffRetry(1000,2))
.build();
curator.start();
this.lockPath = lockPath;
// curator.usingNamespace(lockPath);
mutex = new InterProcessMutex(curator, lockPath);//create the node automatically
// }
}
/**
* get next sequetial number
* @return
* @throws Exception
*/
public int next() throws Exception{
int seq =0;
try {
mutex.acquire();
byte[] seqInByte = curator.getData().forPath(this.lockPath);
seq= Integer.parseInt(new String(seqInByte));
seq++;
curator.setData().forPath(lockPath,new Integer(seq).toString().getBytes());
}catch(Exception e) {
e.printStackTrace();
}finally {
mutex.release();
}
return seq;
}
public static class ClientRunnable implements Runnable{
private static final CyclicBarrier barrier = new CyclicBarrier(5);
boolean interrupted =false;
public void run() {
while(!interrupted) {
DistrSeqGen gen = new DistrSeqGen("localhost:2181","/DistrLock");
try {
barrier.await();
System.out.println(Thread.currentThread().getName() + ":" + gen.next());
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
if(Thread.currentThread().isInterrupted()) {
interrupted = true;
}
}
}
}
public static void main(String[] args) {
// TODO Auto-generated method stub
new Thread(new ClientRunnable()).start();
new Thread(new ClientRunnable()).start();
new Thread(new ClientRunnable()).start();
new Thread(new ClientRunnable()).start();
new Thread(new ClientRunnable()).start();
}
}