zookeeperTutorial.html 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894
  1. <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
  2. <html>
  3. <head>
  4. <META http-equiv="Content-Type" content="text/html; charset=UTF-8">
  5. <meta content="Apache Forrest" name="Generator">
  6. <meta name="Forrest-version" content="0.9">
  7. <meta name="Forrest-skin-name" content="pelt">
  8. <title>Programming with ZooKeeper - A basic tutorial</title>
  9. <link type="text/css" href="skin/basic.css" rel="stylesheet">
  10. <link media="screen" type="text/css" href="skin/screen.css" rel="stylesheet">
  11. <link media="print" type="text/css" href="skin/print.css" rel="stylesheet">
  12. <link type="text/css" href="skin/profile.css" rel="stylesheet">
  13. <script src="skin/getBlank.js" language="javascript" type="text/javascript"></script><script src="skin/getMenu.js" language="javascript" type="text/javascript"></script><script src="skin/fontsize.js" language="javascript" type="text/javascript"></script>
  14. <link rel="shortcut icon" href="images/favicon.ico">
  15. </head>
  16. <body onload="init()">
  17. <script type="text/javascript">ndeSetTextSize();</script>
  18. <div id="top">
  19. <!--+
  20. |breadtrail
  21. +-->
  22. <div class="breadtrail">
  23. <a href="http://www.apache.org/">Apache</a> &gt; <a href="http://hadoop.apache.org/">Hadoop</a> &gt; <a href="http://hadoop.apache.org/zookeeper/">ZooKeeper</a><script src="skin/breadcrumbs.js" language="JavaScript" type="text/javascript"></script>
  24. </div>
  25. <!--+
  26. |header
  27. +-->
  28. <div class="header">
  29. <!--+
  30. |start group logo
  31. +-->
  32. <div class="grouplogo">
  33. <a href="http://hadoop.apache.org/"><img class="logoImage" alt="Hadoop" src="images/hadoop-logo.jpg" title="Apache Hadoop"></a>
  34. </div>
  35. <!--+
  36. |end group logo
  37. +-->
  38. <!--+
  39. |start Project Logo
  40. +-->
  41. <div class="projectlogo">
  42. <a href="http://hadoop.apache.org/zookeeper/"><img class="logoImage" alt="ZooKeeper" src="images/zookeeper_small.gif" title="ZooKeeper: distributed coordination"></a>
  43. </div>
  44. <!--+
  45. |end Project Logo
  46. +-->
  47. <!--+
  48. |start Search
  49. +-->
  50. <div class="searchbox">
  51. <form action="http://www.google.com/search" method="get" class="roundtopsmall">
  52. <input value="hadoop.apache.org" name="sitesearch" type="hidden"><input onFocus="getBlank (this, 'Search the site with google');" size="25" name="q" id="query" type="text" value="Search the site with google">&nbsp;
  53. <input name="Search" value="Search" type="submit">
  54. </form>
  55. </div>
  56. <!--+
  57. |end search
  58. +-->
  59. <!--+
  60. |start Tabs
  61. +-->
  62. <ul id="tabs">
  63. <li>
  64. <a class="unselected" href="http://hadoop.apache.org/zookeeper/">Project</a>
  65. </li>
  66. <li>
  67. <a class="unselected" href="http://wiki.apache.org/hadoop/ZooKeeper">Wiki</a>
  68. </li>
  69. <li class="current">
  70. <a class="selected" href="index.html">ZooKeeper 3.4 Documentation</a>
  71. </li>
  72. </ul>
  73. <!--+
  74. |end Tabs
  75. +-->
  76. </div>
  77. </div>
  78. <div id="main">
  79. <div id="publishedStrip">
  80. <!--+
  81. |start Subtabs
  82. +-->
  83. <div id="level2tabs"></div>
  84. <!--+
  85. |end Endtabs
  86. +-->
  87. <script type="text/javascript"><!--
  88. document.write("Last Published: " + document.lastModified);
  89. // --></script>
  90. </div>
  91. <!--+
  92. |breadtrail
  93. +-->
  94. <div class="breadtrail">
  95. &nbsp;
  96. </div>
  97. <!--+
  98. |start Menu, mainarea
  99. +-->
  100. <!--+
  101. |start Menu
  102. +-->
  103. <div id="menu">
  104. <div onclick="SwitchMenu('menu_1.1', 'skin/')" id="menu_1.1Title" class="menutitle">Overview</div>
  105. <div id="menu_1.1" class="menuitemgroup">
  106. <div class="menuitem">
  107. <a href="index.html">Welcome</a>
  108. </div>
  109. <div class="menuitem">
  110. <a href="zookeeperOver.html">Overview</a>
  111. </div>
  112. <div class="menuitem">
  113. <a href="zookeeperStarted.html">Getting Started</a>
  114. </div>
  115. <div class="menuitem">
  116. <a href="releasenotes.html">Release Notes</a>
  117. </div>
  118. </div>
  119. <div onclick="SwitchMenu('menu_selected_1.2', 'skin/')" id="menu_selected_1.2Title" class="menutitle" style="background-image: url('skin/images/chapter_open.gif');">Developer</div>
  120. <div id="menu_selected_1.2" class="selectedmenuitemgroup" style="display: block;">
  121. <div class="menuitem">
  122. <a href="api/index.html">API Docs</a>
  123. </div>
  124. <div class="menuitem">
  125. <a href="zookeeperProgrammers.html">Programmer's Guide</a>
  126. </div>
  127. <div class="menuitem">
  128. <a href="javaExample.html">Java Example</a>
  129. </div>
  130. <div class="menupage">
  131. <div class="menupagetitle">Barrier and Queue Tutorial</div>
  132. </div>
  133. <div class="menuitem">
  134. <a href="recipes.html">Recipes</a>
  135. </div>
  136. </div>
  137. <div onclick="SwitchMenu('menu_1.3', 'skin/')" id="menu_1.3Title" class="menutitle">BookKeeper</div>
  138. <div id="menu_1.3" class="menuitemgroup">
  139. <div class="menuitem">
  140. <a href="bookkeeperStarted.html">Getting started</a>
  141. </div>
  142. <div class="menuitem">
  143. <a href="bookkeeperOverview.html">Overview</a>
  144. </div>
  145. <div class="menuitem">
  146. <a href="bookkeeperConfig.html">Setup guide</a>
  147. </div>
  148. <div class="menuitem">
  149. <a href="bookkeeperProgrammer.html">Programmer's guide</a>
  150. </div>
  151. </div>
  152. <div onclick="SwitchMenu('menu_1.4', 'skin/')" id="menu_1.4Title" class="menutitle">Admin &amp; Ops</div>
  153. <div id="menu_1.4" class="menuitemgroup">
  154. <div class="menuitem">
  155. <a href="zookeeperAdmin.html">Administrator's Guide</a>
  156. </div>
  157. <div class="menuitem">
  158. <a href="zookeeperQuotas.html">Quota Guide</a>
  159. </div>
  160. <div class="menuitem">
  161. <a href="zookeeperJMX.html">JMX</a>
  162. </div>
  163. <div class="menuitem">
  164. <a href="zookeeperObservers.html">Observers Guide</a>
  165. </div>
  166. </div>
  167. <div onclick="SwitchMenu('menu_1.5', 'skin/')" id="menu_1.5Title" class="menutitle">Contributor</div>
  168. <div id="menu_1.5" class="menuitemgroup">
  169. <div class="menuitem">
  170. <a href="zookeeperInternals.html">ZooKeeper Internals</a>
  171. </div>
  172. </div>
  173. <div onclick="SwitchMenu('menu_1.6', 'skin/')" id="menu_1.6Title" class="menutitle">Miscellaneous</div>
  174. <div id="menu_1.6" class="menuitemgroup">
  175. <div class="menuitem">
  176. <a href="http://wiki.apache.org/hadoop/ZooKeeper">Wiki</a>
  177. </div>
  178. <div class="menuitem">
  179. <a href="http://wiki.apache.org/hadoop/ZooKeeper/FAQ">FAQ</a>
  180. </div>
  181. <div class="menuitem">
  182. <a href="http://hadoop.apache.org/zookeeper/mailing_lists.html">Mailing Lists</a>
  183. </div>
  184. </div>
  185. <div id="credit"></div>
  186. <div id="roundbottom">
  187. <img style="display: none" class="corner" height="15" width="15" alt="" src="skin/images/rc-b-l-15-1body-2menu-3menu.png"></div>
  188. <!--+
  189. |alternative credits
  190. +-->
  191. <div id="credit2"></div>
  192. </div>
  193. <!--+
  194. |end Menu
  195. +-->
  196. <!--+
  197. |start content
  198. +-->
  199. <div id="content">
  200. <div title="Portable Document Format" class="pdflink">
  201. <a class="dida" href="zookeeperTutorial.pdf"><img alt="PDF -icon" src="skin/images/pdfdoc.gif" class="skin"><br>
  202. PDF</a>
  203. </div>
  204. <h1>Programming with ZooKeeper - A basic tutorial</h1>
  205. <div id="front-matter">
  206. <div id="minitoc-area">
  207. <ul class="minitoc">
  208. <li>
  209. <a href="#ch_Introduction">Introduction</a>
  210. </li>
  211. <li>
  212. <a href="#sc_barriers">Barriers</a>
  213. </li>
  214. <li>
  215. <a href="#sc_producerConsumerQueues">Producer-Consumer Queues</a>
  216. </li>
  217. <li>
  218. <a href="#sc_sourceListing">Complete Source Listing</a>
  219. </li>
  220. </ul>
  221. </div>
  222. </div>
  223. <a name="ch_Introduction"></a>
  224. <h2 class="h3">Introduction</h2>
  225. <div class="section">
  226. <p>In this tutorial, we show simple implementations of barriers and
  227. producer-consumer queues using ZooKeeper. We call the respective classes Barrier and Queue.
  228. These examples assume that you have at least one ZooKeeper server running.</p>
  229. <p>Both primitives use the following common excerpt of code:</p>
  230. <pre class="code">
  231. static ZooKeeper zk = null;
  232. static Integer mutex;
  233. String root;
  234. SyncPrimitive(String address) {
  235. if(zk == null){
  236. try {
  237. System.out.println("Starting ZK:");
  238. zk = new ZooKeeper(address, 3000, this);
  239. mutex = new Integer(-1);
  240. System.out.println("Finished starting ZK: " + zk);
  241. } catch (IOException e) {
  242. System.out.println(e.toString());
  243. zk = null;
  244. }
  245. }
  246. }
  247. synchronized public void process(WatchedEvent event) {
  248. synchronized (mutex) {
  249. mutex.notify();
  250. }
  251. }
  252. </pre>
  253. <p>Both classes extend SyncPrimitive. In this way, we execute steps that are
  254. common to all primitives in the constructor of SyncPrimitive. To keep the examples
  255. simple, we create a ZooKeeper object the first time we instantiate either a barrier
  256. object or a queue object, and we declare a static variable that is a reference
  257. to this object. The subsequent instances of Barrier and Queue check whether a
  258. ZooKeeper object exists. Alternatively, we could have the application creating a
  259. ZooKeeper object and passing it to the constructor of Barrier and Queue.</p>
  260. <p>
  261. We use the process() method to process notifications triggered due to watches.
  262. In the following discussion, we present code that sets watches. A watch is internal
  263. structure that enables ZooKeeper to notify a client of a change to a node. For example,
  264. if a client is waiting for other clients to leave a barrier, then it can set a watch and
  265. wait for modifications to a particular node, which can indicate that it is the end of the wait.
  266. This point becomes clear once we go over the examples.
  267. </p>
  268. </div>
  269. <a name="sc_barriers"></a>
  270. <h2 class="h3">Barriers</h2>
  271. <div class="section">
  272. <p>
  273. A barrier is a primitive that enables a group of processes to synchronize the
  274. beginning and the end of a computation. The general idea of this implementation
  275. is to have a barrier node that serves the purpose of being a parent for individual
  276. process nodes. Suppose that we call the barrier node "/b1". Each process "p" then
  277. creates a node "/b1/p". Once enough processes have created their corresponding
  278. nodes, joined processes can start the computation.
  279. </p>
  280. <p>In this example, each process instantiates a Barrier object, and its constructor takes as parameters:</p>
  281. <ul>
  282. <li>
  283. <p>the address of a ZooKeeper server (e.g., "zoo1.foo.com:2181")</p>
  284. </li>
  285. <li>
  286. <p>the path of the barrier node on ZooKeeper (e.g., "/b1")</p>
  287. </li>
  288. <li>
  289. <p>the size of the group of processes</p>
  290. </li>
  291. </ul>
  292. <p>The constructor of Barrier passes the address of the Zookeeper server to the
  293. constructor of the parent class. The parent class creates a ZooKeeper instance if
  294. one does not exist. The constructor of Barrier then creates a
  295. barrier node on ZooKeeper, which is the parent node of all process nodes, and
  296. we call root (<strong>Note:</strong> This is not the ZooKeeper root "/").</p>
  297. <pre class="code">
  298. /**
  299. * Barrier constructor
  300. *
  301. * @param address
  302. * @param root
  303. * @param size
  304. */
  305. Barrier(String address, String root, int size) {
  306. super(address);
  307. this.root = root;
  308. this.size = size;
  309. // Create barrier node
  310. if (zk != null) {
  311. try {
  312. Stat s = zk.exists(root, false);
  313. if (s == null) {
  314. zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
  315. CreateMode.PERSISTENT);
  316. }
  317. } catch (KeeperException e) {
  318. System.out
  319. .println("Keeper exception when instantiating queue: "
  320. + e.toString());
  321. } catch (InterruptedException e) {
  322. System.out.println("Interrupted exception");
  323. }
  324. }
  325. // My node name
  326. try {
  327. name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
  328. } catch (UnknownHostException e) {
  329. System.out.println(e.toString());
  330. }
  331. }
  332. </pre>
  333. <p>
  334. To enter the barrier, a process calls enter(). The process creates a node under
  335. the root to represent it, using its host name to form the node name. It then wait
  336. until enough processes have entered the barrier. A process does it by checking
  337. the number of children the root node has with "getChildren()", and waiting for
  338. notifications in the case it does not have enough. To receive a notification when
  339. there is a change to the root node, a process has to set a watch, and does it
  340. through the call to "getChildren()". In the code, we have that "getChildren()"
  341. has two parameters. The first one states the node to read from, and the second is
  342. a boolean flag that enables the process to set a watch. In the code the flag is true.
  343. </p>
  344. <pre class="code">
  345. /**
  346. * Join barrier
  347. *
  348. * @return
  349. * @throws KeeperException
  350. * @throws InterruptedException
  351. */
  352. boolean enter() throws KeeperException, InterruptedException{
  353. zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
  354. CreateMode.EPHEMERAL_SEQUENTIAL);
  355. while (true) {
  356. synchronized (mutex) {
  357. List&lt;String&gt; list = zk.getChildren(root, true);
  358. if (list.size() &lt; size) {
  359. mutex.wait();
  360. } else {
  361. return true;
  362. }
  363. }
  364. }
  365. }
  366. </pre>
  367. <p>
  368. Note that enter() throws both KeeperException and InterruptedException, so it is
  369. the reponsability of the application to catch and handle such exceptions.</p>
  370. <p>
  371. Once the computation is finished, a process calls leave() to leave the barrier.
  372. First it deletes its corresponding node, and then it gets the children of the root
  373. node. If there is at least one child, then it waits for a notification (obs: note
  374. that the second parameter of the call to getChildren() is true, meaning that
  375. ZooKeeper has to set a watch on the the root node). Upon reception of a notification,
  376. it checks once more whether the root node has any child.</p>
  377. <pre class="code">
  378. /**
  379. * Wait until all reach barrier
  380. *
  381. * @return
  382. * @throws KeeperException
  383. * @throws InterruptedException
  384. */
  385. boolean leave() throws KeeperException, InterruptedException{
  386. zk.delete(root + "/" + name, 0);
  387. while (true) {
  388. synchronized (mutex) {
  389. List&lt;String&gt; list = zk.getChildren(root, true);
  390. if (list.size() &gt; 0) {
  391. mutex.wait();
  392. } else {
  393. return true;
  394. }
  395. }
  396. }
  397. }
  398. }
  399. </pre>
  400. </div>
  401. <a name="sc_producerConsumerQueues"></a>
  402. <h2 class="h3">Producer-Consumer Queues</h2>
  403. <div class="section">
  404. <p>
  405. A producer-consumer queue is a distributed data estructure thata group of processes
  406. use to generate and consume items. Producer processes create new elements and add
  407. them to the queue. Consumer processes remove elements from the list, and process them.
  408. In this implementation, the elements are simple integers. The queue is represented
  409. by a root node, and to add an element to the queue, a producer process creates a new node,
  410. a child of the root node.
  411. </p>
  412. <p>
  413. The following excerpt of code corresponds to the constructor of the object. As
  414. with Barrier objects, it first calls the constructor of the parent class, SyncPrimitive,
  415. that creates a ZooKeeper object if one doesn't exist. It then verifies if the root
  416. node of the queue exists, and creates if it doesn't.
  417. </p>
  418. <pre class="code">
  419. /**
  420. * Constructor of producer-consumer queue
  421. *
  422. * @param address
  423. * @param name
  424. */
  425. Queue(String address, String name) {
  426. super(address);
  427. this.root = name;
  428. // Create ZK node name
  429. if (zk != null) {
  430. try {
  431. Stat s = zk.exists(root, false);
  432. if (s == null) {
  433. zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
  434. CreateMode.PERSISTENT);
  435. }
  436. } catch (KeeperException e) {
  437. System.out
  438. .println("Keeper exception when instantiating queue: "
  439. + e.toString());
  440. } catch (InterruptedException e) {
  441. System.out.println("Interrupted exception");
  442. }
  443. }
  444. }
  445. </pre>
  446. <p>
  447. A producer process calls "produce()" to add an element to the queue, and passes
  448. an integer as an argument. To add an element to the queue, the method creates a
  449. new node using "create()", and uses the SEQUENCE flag to instruct ZooKeeper to
  450. append the value of the sequencer counter associated to the root node. In this way,
  451. we impose a total order on the elements of the queue, thus guaranteeing that the
  452. oldest element of the queue is the next one consumed.
  453. </p>
  454. <pre class="code">
  455. /**
  456. * Add element to the queue.
  457. *
  458. * @param i
  459. * @return
  460. */
  461. boolean produce(int i) throws KeeperException, InterruptedException{
  462. ByteBuffer b = ByteBuffer.allocate(4);
  463. byte[] value;
  464. // Add child with value i
  465. b.putInt(i);
  466. value = b.array();
  467. zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,
  468. CreateMode.PERSISTENT_SEQUENTIAL);
  469. return true;
  470. }
  471. </pre>
  472. <p>
  473. To consume an element, a consumer process obtains the children of the root node,
  474. reads the node with smallest counter value, and returns the element. Note that
  475. if there is a conflict, then one of the two contending processes won't be able to
  476. delete the node and the delete operation will throw an exception.</p>
  477. <p>
  478. A call to getChildren() returns the list of children in lexicographic order.
  479. As lexicographic order does not necessary follow the numerical order of the counter
  480. values, we need to decide which element is the smallest. To decide which one has
  481. the smallest counter value, we traverse the list, and remove the prefix "element"
  482. from each one.</p>
  483. <pre class="code">
  484. /**
  485. * Remove first element from the queue.
  486. *
  487. * @return
  488. * @throws KeeperException
  489. * @throws InterruptedException
  490. */
  491. int consume() throws KeeperException, InterruptedException{
  492. int retvalue = -1;
  493. Stat stat = null;
  494. // Get the first element available
  495. while (true) {
  496. synchronized (mutex) {
  497. List&lt;String&gt; list = zk.getChildren(root, true);
  498. if (list.size() == 0) {
  499. System.out.println("Going to wait");
  500. mutex.wait();
  501. } else {
  502. Integer min = new Integer(list.get(0).substring(7));
  503. for(String s : list){
  504. Integer tempValue = new Integer(s.substring(7));
  505. //System.out.println("Temporary value: " + tempValue);
  506. if(tempValue &lt; min) min = tempValue;
  507. }
  508. System.out.println("Temporary value: " + root + "/element" + min);
  509. byte[] b = zk.getData(root + "/element" + min,
  510. false, stat);
  511. zk.delete(root + "/element" + min, 0);
  512. ByteBuffer buffer = ByteBuffer.wrap(b);
  513. retvalue = buffer.getInt();
  514. return retvalue;
  515. }
  516. }
  517. }
  518. }
  519. }
  520. </pre>
  521. </div>
  522. <a name="sc_sourceListing"></a>
  523. <h2 class="h3">Complete Source Listing</h2>
  524. <div class="section">
  525. <div class="note example">
  526. <div class="label">SyncPrimitive.Java</div>
  527. <div class="content">
  528. <title>SyncPrimitive.Java</title>
  529. <pre class="code">
  530. import java.io.IOException;
  531. import java.net.InetAddress;
  532. import java.net.UnknownHostException;
  533. import java.nio.ByteBuffer;
  534. import java.util.List;
  535. import java.util.Random;
  536. import org.apache.zookeeper.CreateMode;
  537. import org.apache.zookeeper.KeeperException;
  538. import org.apache.zookeeper.WatchedEvent;
  539. import org.apache.zookeeper.Watcher;
  540. import org.apache.zookeeper.ZooKeeper;
  541. import org.apache.zookeeper.ZooDefs.Ids;
  542. import org.apache.zookeeper.data.Stat;
  543. public class SyncPrimitive implements Watcher {
  544. static ZooKeeper zk = null;
  545. static Integer mutex;
  546. String root;
  547. SyncPrimitive(String address) {
  548. if(zk == null){
  549. try {
  550. System.out.println("Starting ZK:");
  551. zk = new ZooKeeper(address, 3000, this);
  552. mutex = new Integer(-1);
  553. System.out.println("Finished starting ZK: " + zk);
  554. } catch (IOException e) {
  555. System.out.println(e.toString());
  556. zk = null;
  557. }
  558. }
  559. //else mutex = new Integer(-1);
  560. }
  561. synchronized public void process(WatchedEvent event) {
  562. synchronized (mutex) {
  563. //System.out.println("Process: " + event.getType());
  564. mutex.notify();
  565. }
  566. }
  567. /**
  568. * Barrier
  569. */
  570. static public class Barrier extends SyncPrimitive {
  571. int size;
  572. String name;
  573. /**
  574. * Barrier constructor
  575. *
  576. * @param address
  577. * @param root
  578. * @param size
  579. */
  580. Barrier(String address, String root, int size) {
  581. super(address);
  582. this.root = root;
  583. this.size = size;
  584. // Create barrier node
  585. if (zk != null) {
  586. try {
  587. Stat s = zk.exists(root, false);
  588. if (s == null) {
  589. zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
  590. CreateMode.PERSISTENT);
  591. }
  592. } catch (KeeperException e) {
  593. System.out
  594. .println("Keeper exception when instantiating queue: "
  595. + e.toString());
  596. } catch (InterruptedException e) {
  597. System.out.println("Interrupted exception");
  598. }
  599. }
  600. // My node name
  601. try {
  602. name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
  603. } catch (UnknownHostException e) {
  604. System.out.println(e.toString());
  605. }
  606. }
  607. /**
  608. * Join barrier
  609. *
  610. * @return
  611. * @throws KeeperException
  612. * @throws InterruptedException
  613. */
  614. boolean enter() throws KeeperException, InterruptedException{
  615. zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,
  616. CreateMode.EPHEMERAL_SEQUENTIAL);
  617. while (true) {
  618. synchronized (mutex) {
  619. List&lt;String&gt; list = zk.getChildren(root, true);
  620. if (list.size() &lt; size) {
  621. mutex.wait();
  622. } else {
  623. return true;
  624. }
  625. }
  626. }
  627. }
  628. /**
  629. * Wait until all reach barrier
  630. *
  631. * @return
  632. * @throws KeeperException
  633. * @throws InterruptedException
  634. */
  635. boolean leave() throws KeeperException, InterruptedException{
  636. zk.delete(root + "/" + name, 0);
  637. while (true) {
  638. synchronized (mutex) {
  639. List&lt;String&gt; list = zk.getChildren(root, true);
  640. if (list.size() &gt; 0) {
  641. mutex.wait();
  642. } else {
  643. return true;
  644. }
  645. }
  646. }
  647. }
  648. }
  649. /**
  650. * Producer-Consumer queue
  651. */
  652. static public class Queue extends SyncPrimitive {
  653. /**
  654. * Constructor of producer-consumer queue
  655. *
  656. * @param address
  657. * @param name
  658. */
  659. Queue(String address, String name) {
  660. super(address);
  661. this.root = name;
  662. // Create ZK node name
  663. if (zk != null) {
  664. try {
  665. Stat s = zk.exists(root, false);
  666. if (s == null) {
  667. zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,
  668. CreateMode.PERSISTENT);
  669. }
  670. } catch (KeeperException e) {
  671. System.out
  672. .println("Keeper exception when instantiating queue: "
  673. + e.toString());
  674. } catch (InterruptedException e) {
  675. System.out.println("Interrupted exception");
  676. }
  677. }
  678. }
  679. /**
  680. * Add element to the queue.
  681. *
  682. * @param i
  683. * @return
  684. */
  685. boolean produce(int i) throws KeeperException, InterruptedException{
  686. ByteBuffer b = ByteBuffer.allocate(4);
  687. byte[] value;
  688. // Add child with value i
  689. b.putInt(i);
  690. value = b.array();
  691. zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,
  692. CreateMode.PERSISTENT_SEQUENTIAL);
  693. return true;
  694. }
  695. /**
  696. * Remove first element from the queue.
  697. *
  698. * @return
  699. * @throws KeeperException
  700. * @throws InterruptedException
  701. */
  702. int consume() throws KeeperException, InterruptedException{
  703. int retvalue = -1;
  704. Stat stat = null;
  705. // Get the first element available
  706. while (true) {
  707. synchronized (mutex) {
  708. List&lt;String&gt; list = zk.getChildren(root, true);
  709. if (list.size() == 0) {
  710. System.out.println("Going to wait");
  711. mutex.wait();
  712. } else {
  713. Integer min = new Integer(list.get(0).substring(7));
  714. for(String s : list){
  715. Integer tempValue = new Integer(s.substring(7));
  716. //System.out.println("Temporary value: " + tempValue);
  717. if(tempValue &lt; min) min = tempValue;
  718. }
  719. System.out.println("Temporary value: " + root + "/element" + min);
  720. byte[] b = zk.getData(root + "/element" + min,
  721. false, stat);
  722. zk.delete(root + "/element" + min, 0);
  723. ByteBuffer buffer = ByteBuffer.wrap(b);
  724. retvalue = buffer.getInt();
  725. return retvalue;
  726. }
  727. }
  728. }
  729. }
  730. }
  731. public static void main(String args[]) {
  732. if (args[0].equals("qTest"))
  733. queueTest(args);
  734. else
  735. barrierTest(args);
  736. }
  737. public static void queueTest(String args[]) {
  738. Queue q = new Queue(args[1], "/app1");
  739. System.out.println("Input: " + args[1]);
  740. int i;
  741. Integer max = new Integer(args[2]);
  742. if (args[3].equals("p")) {
  743. System.out.println("Producer");
  744. for (i = 0; i &lt; max; i++)
  745. try{
  746. q.produce(10 + i);
  747. } catch (KeeperException e){
  748. } catch (InterruptedException e){
  749. }
  750. } else {
  751. System.out.println("Consumer");
  752. for (i = 0; i &lt; max; i++) {
  753. try{
  754. int r = q.consume();
  755. System.out.println("Item: " + r);
  756. } catch (KeeperException e){
  757. i--;
  758. } catch (InterruptedException e){
  759. }
  760. }
  761. }
  762. }
  763. public static void barrierTest(String args[]) {
  764. Barrier b = new Barrier(args[1], "/b1", new Integer(args[2]));
  765. try{
  766. boolean flag = b.enter();
  767. System.out.println("Entered barrier: " + args[2]);
  768. if(!flag) System.out.println("Error when entering the barrier");
  769. } catch (KeeperException e){
  770. } catch (InterruptedException e){
  771. }
  772. // Generate random integer
  773. Random rand = new Random();
  774. int r = rand.nextInt(100);
  775. // Loop for rand iterations
  776. for (int i = 0; i &lt; r; i++) {
  777. try {
  778. Thread.sleep(100);
  779. } catch (InterruptedException e) {
  780. }
  781. }
  782. try{
  783. b.leave();
  784. } catch (KeeperException e){
  785. } catch (InterruptedException e){
  786. }
  787. System.out.println("Left barrier");
  788. }
  789. }
  790. </pre>
  791. </div>
  792. </div>
  793. </div>
  794. <p align="right">
  795. <font size="-2"></font>
  796. </p>
  797. </div>
  798. <!--+
  799. |end content
  800. +-->
  801. <div class="clearboth">&nbsp;</div>
  802. </div>
  803. <div id="footer">
  804. <!--+
  805. |start bottomstrip
  806. +-->
  807. <div class="lastmodified">
  808. <script type="text/javascript"><!--
  809. document.write("Last Published: " + document.lastModified);
  810. // --></script>
  811. </div>
  812. <div class="copyright">
  813. Copyright &copy;
  814. 2008 <a href="http://www.apache.org/licenses/">The Apache Software Foundation.</a>
  815. </div>
  816. <!--+
  817. |end bottomstrip
  818. +-->
  819. </div>
  820. </body>
  821. </html>