分佈式隊列ZooKeeper的實現

時間 2019-11-09

標籤分佈式隊列 zookeeper 實現欄目系統架構简体版

原文原文鏈接

1、背景

　　有一些時候，多個團隊須要共同完成一個任務，好比，A團隊將Hadoop集羣計算的結果交給B團隊繼續計算，B完成了本身任務再交給C團隊繼續作。這就有點像業務系統的工做流同樣，一環一環地傳下html

去，直到最後一部分完成。在業務系統中，咱們常常會用SOA的架構來解決這種問題，每一個團隊在ESB（企業服務股總線）服務器上部署本身的服務，而後經過消息中間件完成調度任務。對亍分步式的多個java

Hadoop集羣系統的協做，一樣能夠用這種架構來作只要把消息中間件引擎換成支持分步式的消息中間件的引擎就好了。node

　　本文樓主將使用zookeeper作爲分步式消息中間件構造一個大型超市的部分數據計算模型來完成各個區域利潤計算的業務需求。git

　　因爲採購和銷售分別是由不一樣廠商進行的軟件開發和維護，並且業務往來也在不一樣的城市和地區。因此在每個月底結算時，工做量都特別大。好比，計算利潤表：當月利潤 = 當月銷售金額 - 當月採購github

額 - 當月其餘支出（樓主只是粗略計算）。若是採購系統是單獨的系統，銷售是另外單獨的系統，及以其餘幾十個大大小小的系統，如何能讓多個系統，配合起來完成該需求？數據庫

2、系統構思

　　樓主基於zookeeper來構建一個分步式隊列的應用，來解決上面的功能需求。排除了ESB的部分，只保留zookeeper進行實現。apache

　　採購數據：海量數據，基於Hadoop存儲和分析（樓主環境有限，只使用了不多的數據）
　　銷售數據：海量數據，基於Hadoop存儲和分析（樓主環境有限，只使用了不多的數據）
　　其餘費用支出：爲少許數據，基於文件或數據庫存儲和分析

　　咱們設計一個同步隊列，這個隊列有3個條件節點，分別對應採購(purchase)，銷售 (sell)，其餘費用(other)3個部分。當3個節點都被建立後，程序會自動觸發計算利潤，幵建立利潤(profit)節點。上面3個節點的建立，無順序要求。每一個節點只能被建立一次。

服務器

　　Hadoop mapreduce1,Hadoop mapreduce2 是2個獨立的Hadoop集羣應用。 Java App 是2個獨立的Java應用。ZooKeeper集羣的有3個節點。架構

/queue，是znode的隊列目錄，假設隊列長度爲3
/queue/purchase，是znode隊列中，1號排對者，由Hadoop mapreduce1提交，用於統計採購金額
/queue/sell，是znode隊列中，2號排對者，由Hadoop mapreduce2提交，用於統計銷售金額
/queue/other，是znode隊列中，3號排對者，由Java App提交，用於統計其餘費用支出金額
/queue/profit，當znode隊列中滿了，觸發建立利潤節點。

　　當/qeueu/profit被建立後，利潤java app被啓動，全部zookeeper的鏈接通知同步程序(紅色線)，隊列已完成，全部程序結束。app

3、環境準備

　　1）hadoop集羣。樓主用的6個節點的hadoop2.7.3集羣，各位同窗能夠根據本身的實際狀況進行搭建，但至少須要1臺僞分佈式的。（參考http://www.cnblogs.com/qq503665965/p/6790580.html）

　　2）zookeeper集羣。至少三個節點。安裝參考樓主這篇文章（http://www.cnblogs.com/qq503665965/p/6790580.html）

　　3）java開發環境。

4、mapreduce及java app程序

　　計算採購金額：

  1 package zkqueue;
  2 import java.io.IOException;
  3 import java.util.HashMap;
  4 import java.util.Map;
  5 import java.util.regex.Pattern;
  6 
  7 import org.apache.hadoop.fs.Path;
  8 import org.apache.hadoop.io.IntWritable;
  9 import org.apache.hadoop.io.LongWritable;
 10 import org.apache.hadoop.io.Text;
 11 import org.apache.hadoop.mapred.JobConf;
 12 import org.apache.hadoop.mapreduce.Job;
 13 import org.apache.hadoop.mapreduce.Mapper;
 14 import org.apache.hadoop.mapreduce.Reducer;
 15 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 16 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
 17 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 18 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 19 
 20 
 21 /**
 22  * 採購金額計算
 23  * @author Jon_China
 24  *
 25  */
 26 public class Purchase {
 27 
 28     public static final String HDFS = "hdfs://192.168.8.101:9000";
 29     public static final Pattern DELIMITER = Pattern.compile("[\t,]");
 30 
 31     public static class PurchaseMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
 32 
 33         private String month = "2017-01";
 34         private Text k = new Text(month);
 35         private IntWritable v = new IntWritable();
 36         private int money = 0;
 37 
 38         public void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException {
 39             System.out.println(values.toString());
 40             String[] tokens = DELIMITER.split(values.toString());//拆分源數據
 41             if (tokens[3].startsWith(month)) {// 過濾1月份數據
 42                 money = Integer.parseInt(tokens[1]) * Integer.parseInt(tokens[2]);//計算
 43                 v.set(money);
 44                 context.write(k, v);
 45             }
 46         }
 47     }
 48 
 49     public static class PurchaseReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
 50         private IntWritable v = new IntWritable();
 51         private int money = 0;
 52 
 53         @Override
 54         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
 55             for (IntWritable line : values) {
 56                 money += line.get();
 57             }
 58             v.set(money);
 59             context.write(null, v);
 60             System.out.println("Output:" + key + "," + money);
 61         }
 62 
 63     }
 64 
 65     public static void run(Map<String, String> path) throws IOException, InterruptedException, ClassNotFoundException {
 66         JobConf conf = config();
 67         String local_data = path.get("purchase");
 68         String input = path.get("input");
 69         String output = path.get("output");
 70 
 71         
 72         HdfsDAO hdfs = new HdfsDAO(HDFS, conf);
 73         hdfs.rmr(input);
 74         hdfs.mkdirs(input);
 75         hdfs.copyFile(local_data, input);
 76 
 77         Job job = Job.getInstance(conf);
 78         job.setJarByClass(Purchase.class);
 79 
 80         job.setOutputKeyClass(Text.class);
 81         job.setOutputValueClass(IntWritable.class);
 82 
 83         job.setMapperClass(PurchaseMapper.class);
 84         job.setReducerClass(PurchaseReducer.class);
 85 
 86         job.setInputFormatClass(TextInputFormat.class);
 87         job.setOutputFormatClass(TextOutputFormat.class);
 88 
 89         FileInputFormat.setInputPaths(job, new Path(input));
 90         FileOutputFormat.setOutputPath(job, new Path(output));
 91 
 92         job.waitForCompletion(true);
 93     }
 94 
 95     public static JobConf config() {
 96         JobConf conf = new JobConf(Purchase.class);
 97         conf.setJobName("purchase");
 98         conf.addResource("classpath:/hadoop/core-site.xml");
 99         conf.addResource("classpath:/hadoop/hdfs-site.xml");
100         conf.addResource("classpath:/hadoop/mapred-site.xml");
101         conf.addResource("classpath:/hadoop/yarn-site.xml");
102         return conf;
103     }
104     
105     public static Map<String,String> path(){
106         Map<String, String> path = new HashMap<String, String>();
107         path.put("purchase", Purchase.class.getClassLoader().getResource("logfile/biz/purchase.csv").getPath());// 源文件數據
108         path.put("input", HDFS + "/user/hdfs/biz/purchase");//hdfs存儲路徑
109         path.put("output", HDFS + "/user/hdfs/biz/purchase/output"); //hdfs輸出路徑
110         return path;
111     }
112 
113     public static void main(String[] args) throws Exception {
114         run(path());
115     }
116 
117 }

　　銷售數據計算：

  1 package zkqueue;
  2 
  3 import java.io.IOException;
  4 import java.util.HashMap;
  5 import java.util.Map;
  6 import java.util.regex.Pattern;
  7 
  8 import org.apache.hadoop.fs.Path;
  9 import org.apache.hadoop.io.IntWritable;
 10 import org.apache.hadoop.io.LongWritable;
 11 import org.apache.hadoop.io.Text;
 12 import org.apache.hadoop.mapred.JobConf;
 13 import org.apache.hadoop.mapreduce.Job;
 14 import org.apache.hadoop.mapreduce.Mapper;
 15 import org.apache.hadoop.mapreduce.Reducer;
 16 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 17 import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
 18 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
 19 import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
 20 
 21 /**
 22  * 銷售數據計算
 23  * @author Jon_China
 24  *
 25  */
 26 public class Sell {
 27 
 28     public static final String HDFS = "hdfs://192.168.8.101:9000";
 29     public static final Pattern DELIMITER = Pattern.compile("[\t,]");
 30 
 31     public static class SellMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
 32 
 33         private String month = "2013-01";
 34         private Text k = new Text(month);
 35         private IntWritable v = new IntWritable();
 36         private int money = 0;
 37 
 38         public void map(LongWritable key, Text values, Context context) throws IOException, InterruptedException {
 39             System.out.println(values.toString());
 40             String[] tokens = DELIMITER.split(values.toString());
 41             if (tokens[3].startsWith(month)) {// 1月的數據
 42                 money = Integer.parseInt(tokens[1]) * Integer.parseInt(tokens[2]);//單價*數量
 43                 v.set(money);
 44                 context.write(k, v);
 45             }
 46         }
 47     }
 48 
 49     public static class SellReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
 50         private IntWritable v = new IntWritable();
 51         private int money = 0;
 52 
 53         @Override
 54         public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
 55             for (IntWritable line : values) {
 56                 money += line.get();
 57             }
 58             v.set(money);
 59             context.write(null, v);
 60             System.out.println("Output:" + key + "," + money);
 61         }
 62 
 63     }
 64 
 65     public static void run(Map<String, String> path) throws IOException, InterruptedException, ClassNotFoundException {
 66         JobConf conf = config();
 67         String local_data = path.get("sell");
 68         String input = path.get("input");
 69         String output = path.get("output");
 70 
 71         // 初始化sell
 72         HdfsDAO hdfs = new HdfsDAO(HDFS, conf);
 73         hdfs.rmr(input);
 74         hdfs.mkdirs(input);
 75         hdfs.copyFile(local_data, input);
 76 
 77         Job job = Job.getInstance(conf);
 78         job.setJarByClass(Sell.class);
 79 
 80         job.setOutputKeyClass(Text.class);
 81         job.setOutputValueClass(IntWritable.class);
 82 
 83         job.setMapperClass(SellMapper.class);
 84         job.setReducerClass(SellReducer.class);
 85 
 86         job.setInputFormatClass(TextInputFormat.class);
 87         job.setOutputFormatClass(TextOutputFormat.class);
 88 
 89         FileInputFormat.setInputPaths(job, new Path(input));
 90         FileOutputFormat.setOutputPath(job, new Path(output));
 91 
 92         job.waitForCompletion(true);
 93     }
 94 
 95     public static JobConf config() {// Hadoop集羣的遠程配置信息
 96         JobConf conf = new JobConf(Purchase.class);
 97         conf.setJobName("purchase");
 98         conf.addResource("classpath:/hadoop/core-site.xml");
 99         conf.addResource("classpath:/hadoop/hdfs-site.xml");
100         conf.addResource("classpath:/hadoop/mapred-site.xml");
101         conf.addResource("classpath:/hadoop/yarn-site.xml");
102         return conf;
103     }
104     
105     public static Map<String,String> path(){
106         Map<String, String> path = new HashMap<String, String>();
107         path.put("sell", Sell.class.getClassLoader().getResource("logfile/biz/sell.csv").getPath());// 本地的數據文件
108         path.put("input", HDFS + "/user/hdfs/biz/sell");// HDFS的目錄
109         path.put("output", HDFS + "/user/hdfs/biz/sell/output"); // 輸出目錄
110         return path;
111     }
112 
113     public static void main(String[] args) throws Exception {
114         run(path());
115     }
116 
117 }

　　其餘金額計算：

 1 package zkqueue;
 2 
 3 import java.io.BufferedReader;
 4 import java.io.File;
 5 import java.io.FileReader;
 6 import java.io.IOException;
 7 import java.util.regex.Pattern;
 8 
 9 public class Other {
10 
11     public static String file = "/logfile/biz/other.csv";
12     public static final Pattern DELIMITER = Pattern.compile("[\t,]");
13     private static String month = "2017-01";
14 
15     public static void main(String[] args) throws IOException {
16         calcOther(file);
17     }
18 
19     public static int calcOther(String file) throws IOException {
20         int money = 0;
21         BufferedReader br = new BufferedReader(new FileReader(new File(file)));
22 
23         String s = null;
24         while ((s = br.readLine()) != null) {
25             String[] tokens = DELIMITER.split(s);
26             if (tokens[0].startsWith(month)) {// 1月的數據
27                 money += Integer.parseInt(tokens[1]);
28             }
29         }
30         br.close();
31         System.out.println("Output:" + month + "," + money);
32         return money;
33     }
34 }

　　計算利潤：

 1 package zkqueue;
 2 
 3 import java.io.IOException;
 4 
 5 
 6 /**
 7  * 利潤計算
 8  * @author Jon_China
 9  *
10  */
11 public class Profit {
12 
13     public static void main(String[] args) throws Exception {
14         profit();
15     }
16 
17     public static void profit() throws Exception {
18         int sell = getSell();
19         int purchase = getPurchase();
20         int other = getOther();
21         int profit = sell - purchase - other;
22         System.out.printf("profit = sell - purchase - other = %d - %d - %d = %d\n", sell, purchase, other, profit);
23     }
24 
25     public static int getPurchase() throws Exception {
26         HdfsDAO hdfs = new HdfsDAO(Purchase.HDFS, Purchase.config());
27         return Integer.parseInt(hdfs.cat(Purchase.path().get("output") + "/part-r-00000").trim());
28     }
29 
30     public static int getSell() throws Exception {
31         HdfsDAO hdfs = new HdfsDAO(Sell.HDFS, Sell.config());
32         return Integer.parseInt(hdfs.cat(Sell.path().get("output") + "/part-r-00000").trim());
33     }
34 
35     public static int getOther() throws IOException {
36         return Other.calcOther(Other.file);
37     }
38 
39 }

　　zookeeper任務調度：

  1 package zkqueue;
  2 
  3 import java.io.IOException;
  4 import java.util.List;
  5 
  6 import org.apache.zookeeper.CreateMode;
  7 import org.apache.zookeeper.KeeperException;
  8 import org.apache.zookeeper.WatchedEvent;
  9 import org.apache.zookeeper.Watcher;
 10 import org.apache.zookeeper.ZooDefs.Ids;
 11 import org.apache.zookeeper.ZooKeeper;
 12 /**
 13  * 分佈式隊列zookeeper調度
 14  * @author Jon_China
 15  *
 16  */
 17 public class QueueZookeeper {
 18     //設置隊列目錄樹
 19     final public static String QUEUE = "/queue";
 20     final public static String PROFIT = "/queue/profit";
 21     final public static String PURCHASE = "/queue/purchase";
 22     final public static String SELL = "/queue/sell";
 23     final public static String OTHER = "/queue/other";
 24 
 25     public static void main(String[] args) throws Exception {
 26         if (args.length == 0) {
 27             System.out.println("Please start a task:");
 28         } else {
 29             doAction(Integer.parseInt(args[0]));
 30         }
 31     }
 32     public static void doAction(int client) throws Exception {
 33         //zookeeper地址
 34         String host1 = "192.168.8.104:2181";
 35         String host2 = "192.168.8.105:2181";
 36         String host3 = "192.168.8.106:2181";
 37 
 38         ZooKeeper zk = null;
 39         switch (client) {//1,2,3分別將不一樣任務加入隊列
 40         case 1:
 41             zk = connection(host1);
 42             initQueue(zk);
 43             doPurchase(zk);
 44             break;
 45         case 2:
 46             zk = connection(host2);
 47             initQueue(zk);
 48             doSell(zk);
 49             break;
 50         case 3:
 51             zk = connection(host3);
 52             initQueue(zk);
 53             doOther(zk);
 54             break;
 55         }
 56     }
 57 
 58     // 建立一個與服務器的鏈接
 59     public static ZooKeeper connection(String host) throws IOException {
 60         ZooKeeper zk = new ZooKeeper(host, 60000, new Watcher() {
 61             // 監控全部被觸發的事件
 62             public void process(WatchedEvent event) {
 63                 if (event.getType() == Event.EventType.NodeCreated && event.getPath().equals(PROFIT)) {
 64                     System.out.println("Queue has Completed!!!");
 65                 }
 66             }
 67         });
 68         return zk;
 69     }
 70     /**
 71      * 初始化隊列
 72      * @param zk
 73      * @throws KeeperException
 74      * @throws InterruptedException
 75      */
 76     public static void initQueue(ZooKeeper zk) throws KeeperException, InterruptedException {
 77         System.out.println("WATCH => " + PROFIT);
 78         zk.exists(PROFIT, true);
 79 
 80         if (zk.exists(QUEUE, false) == null) {
 81             System.out.println("create " + QUEUE);
 82             zk.create(QUEUE, QUEUE.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 83         } else {
 84             System.out.println(QUEUE + " is exist!");
 85         }
 86     }
 87     /**
 88      * 採購任務
 89      * @param zk
 90      * @throws Exception
 91      */
 92     public static void doPurchase(ZooKeeper zk) throws Exception {
 93         if (zk.exists(PURCHASE, false) == null) {
 94             
 95             Purchase.run(Purchase.path());
 96             
 97             System.out.println("create " + PURCHASE);
 98             zk.create(PURCHASE, PURCHASE.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
 99         } else {
100             System.out.println(PURCHASE + " is exist!");
101         }
102         isCompleted(zk);
103     }
104     /**
105      * 銷售任務
106      * @param zk
107      * @throws Exception
108      */
109     public static void doSell(ZooKeeper zk) throws Exception {
110         if (zk.exists(SELL, false) == null) {
111             
112             Sell.run(Sell.path());
113             
114             System.out.println("create " + SELL);
115             zk.create(SELL, SELL.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
116         } else {
117             System.out.println(SELL + " is exist!");
118         }
119         isCompleted(zk);
120     }
121     /**
122      * 其餘計算任務
123      * @param zk
124      * @throws Exception
125      */
126     public static void doOther(ZooKeeper zk) throws Exception {
127         if (zk.exists(OTHER, false) == null) {
128             
129             Other.calcOther(Other.file);
130             
131             System.out.println("create " + OTHER);
132             zk.create(OTHER, OTHER.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
133         } else {
134             System.out.println(OTHER + " is exist!");
135         }
136         isCompleted(zk);
137     }
138     /**
139      * 檢測完成狀況
140      * @param zk
141      * @throws Exception
142      */
143     public static void isCompleted(ZooKeeper zk) throws Exception {
144         int size = 3;
145         List<String> children = zk.getChildren(QUEUE, true);
146         int length = children.size();
147 
148         System.out.println("Queue Complete:" + length + "/" + size);
149         if (length >= size) {
150             System.out.println("create " + PROFIT);
151             Profit.profit();
152             zk.create(PROFIT, PROFIT.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
153 
154             for (String child : children) {// 清空節點
155                 zk.delete(QUEUE + "/" + child, -1);
156             }
157 
158         }
159     }
160 }