How to store data in zookeeper in a specified format
This article mainly explains "how to store the data in zookeeper according to the specified format". The content in the article is simple and clear, and it is easy to learn and understand. Please follow the editor's train of thought to study and learn "how to store data in zookeeper according to the specified format".
Environment:
Scala version: 2.11.8
Zookeeper version: 3.4.5-cdh6.7.0
Package com.ruozedata.zkimport java.util.concurrent.TimeUnitimport org.apache.curator.framework.CuratorFrameworkFactoryimport org.apache.curator.framework.recipes.locks.InterProcessMuteximport org.apache.curator.retry.ExponentialBackoffRetryimport org.slf4j.LoggerFactoryimport scala.collection.JavaConversions._import scala.collection.mutable/** * Created by ganwei on 2018-08-21 * requires: * 1 to store data in zookeeper through storeOffsets method. * Storage format: * / consumers/G322/offsets/ruoze_offset_topic/partition/0 * / consumers/G322/offsets/ruoze_offset_topic/partition/1 * / consumers/G322/offsets/ruoze_offset_topic/partition/2 * 2 read the stored data through the obtainOffsets method * output format: * topic:ruoze_offset_topic partition:0 Offset:7 * topic:ruoze_offset_topic partition:1 offset:3 * topic:ruoze_offset_topic partition:2 offset:5 * / object ZkConnectApp {val LOG = LoggerFactory.getLogger (ZkConnectApp.getClass) val client = {val client = CuratorFrameworkFactory .builder .connectString ("172.16.100.31purl 2181") .retryPolicy (new ExponentialBackoffRetry (1000) ) .namespace ("consumers") .build () client.start () client} def lock (path: String) (body: = > Unit) {val lock = new InterProcessMutex (client, path) lock.acquire () try {body} finally {lock.release ()} def tryDo (path: String) (body: = > Unit): Boolean = {val lock = new InterProcessMutex (client) Path) if (! lock.acquire (10, TimeUnit.SECONDS)) {LOG.info (s "cannot acquire lock {$path} There are already tasks running. This task exits ") return false} try {LOG.info (" allowed to run ") body true} finally {lock.release () LOG.info (s" release lock {$path} ")}} / / zookeeper creation path def ensurePathExists (path: String): Unit = {if (client.checkExists (). ForPath (path) = = null) {client .create () .creatingParentsIfNeeded () .forPath (path)}} / * OffsetRange class definition (offset object) * used to store offset * / case class OffsetRange (val topic:String / / theme val partition:Int, / / partition val fromOffset:Long / / start offset val utilOffset:Long / / end offset) / * * zookeeper method for storing offset * write format: * / consumers/G322/offsets/ruoze_offset_topic/partition/0 * / consumers/G322/offsets/ruoze_offset_topic/partition/1 * / consumers/G322/ Offsets/ruoze_offset_topic/partition/2 * @ param OffsetsRanges * @ param groupName * / def storeOffsets (OffsetsRanges: array [OffsetRange] GroupName:String) = {val offsetRootPath = s "/" + groupName if (client.checkExists (). ForPath (offsetRootPath) = = null) {client.create (). CreatingParentsIfNeeded (). ForPath (offsetRootPath)} for (els)
