摘要：前面的文章將做爬蟲(chóng)的基礎(chǔ)一直到部署都梳理了一遍，現(xiàn)在來(lái)看一下分布式的處理為什么需要分布式需要抓取的不同數(shù)據(jù)有很多，會(huì)同時(shí)開(kāi)啟無(wú)頭瀏覽器去抓取，然后獲取到數(shù)據(jù)后又無(wú)厘頭的一股腦擠進(jìn)數(shù)據(jù)庫(kù)無(wú)法保證同一時(shí)刻需要的數(shù)據(jù)只有一個(gè)操作在進(jìn)行分布式選擇因

   1. 需要抓取的不同數(shù)據(jù)有很多，會(huì)同時(shí)開(kāi)啟無(wú)頭瀏覽器去抓取，然后獲取到數(shù)據(jù)后又無(wú)厘頭的一股腦擠進(jìn)數(shù)據(jù)庫(kù)
   2. 無(wú)法保證同一時(shí)刻需要的數(shù)據(jù)只有一個(gè)操作在進(jìn)行

因?yàn)槭褂玫氖莕ode，所以盡可能的尋找node支持的分布式框架
ZooKeeper 和 RabbitMQ 的思想百度上有好多說(shuō)明，讀者可以自行搜索作更詳細(xì)的了解
node版的zookeeper
node版的RabbitMQ

發(fā)布者，給書(shū)旗起一個(gè)標(biāo)識(shí)為 37 (channel_id)，然后是要抓取書(shū)的書(shū)籍id(channel_book_id)

// 我們以接口的形式接收爬取的參數(shù)    
// 簡(jiǎn)易版請(qǐng)求(除了接收參數(shù)不做任何處理) -> 發(fā)布者
app.get("/v1.0/grasp_book", (req, res, next) => {
  // 抓取時(shí)需要的參數(shù)
  if (!req.query.channel_id && !req.query.channel_book_id) {
    res.send({
      code: 403,
      msg: "params error"
    })
    return null;
  }
  // 發(fā)布者
  // 連接rabbitmq
  amqp.connect("amqp://rabbitmq:[email protected]:5672/").then(function(conn) {
    return conn.createChannel().then(function(ch) {
      // 創(chuàng)建 hello 的消息隊(duì)列
      var q = "hello";
      // 解析為json字符串格式作為傳遞的數(shù)據(jù)格式
      var msg = JSON.stringify({
        "channel_id": req.query.channel_id,
        "book_id": req.query.book_id
      })
      // 連接并保持
      var ok = ch.assertQueue(q, {durable: true});
      return ok.then(function(_qok) {
        // 發(fā)送數(shù)據(jù)到消費(fèi)者
        ch.sendToQueue(q, Buffer.from(msg));
        console.log(" [x] Sent "%s"", msg);
        return ch.close();
      });
    }).finally(function() { conn.close(); });
  }).catch(console.warn);
  res.send({
    code: 200,
    msg: "success"
  });
}); 

命令行啟動(dòng)app.js，發(fā)出請(qǐng)求并查看結(jié)果，{} 里面的就是我們發(fā)送出去的數(shù)據(jù)

消費(fèi)者，接收到發(fā)布者傳遞過(guò)來(lái)的數(shù)據(jù)建立消息隊(duì)列，然后用Zookeeper創(chuàng)建臨時(shí)節(jié)點(diǎn)以保持隊(duì)列依次執(zhí)行

var zookeeper = require("node-zookeeper-client");

// 根據(jù)標(biāo)識(shí)動(dòng)態(tài)引入js文件
function moduleCustomize(channel_id) {
    return require(`../${channel_id}.js`)
}

 
var client = zookeeper.createClient("127.0.0.1:2181");

async function sleep(second) {
  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve("sleep")
    }, second)
  })
}

// 連接_zookeeper
client.once("connected", function () {
  console.log("Connected to the server.");
  // 建立連接 rabbitmp
  amqp.connect("amqp://rabbitmq:[email protected]:5672/").then(function(conn) {
    return conn.createChannel().then(function(ch) {
      // 與名為 hello(由發(fā)布者創(chuàng)建) 的消息隊(duì)列建立連接
      var ok = ch.assertQueue("hello", {durable: true});
      // 預(yù)存為1
      ok = ok.then(function() { ch.prefetch(1); });
      ok = ok.then(function() {
        // doWork 回調(diào)函數(shù) -> 執(zhí)行接收到數(shù)據(jù)后的操作
        ch.consume("hello", doWork, {noAck: false});
      });
      return ok;
      // rabbitmq 處理
      function doWork(msg) {
          // 接收到數(shù)據(jù)
          var body = msg.content.toString();
          console.log("[x] Received "%s"", body);
          let _body = JSON.parse(body)
          let channel_book_id = _body["channel_book_id"];
          let channel_id = _body["channel_id"];
          // zookeeper 節(jié)點(diǎn)
          let path = `/${channel_id + "_" + channel_book_id}`
          // 連接_zookeeper 判斷是否存在
          client.exists(path, function (error, stat) {
              if (error) {
                  console.log(error.stack);
                  return;
              }
              if (stat) {
                console.log("Node exists.");
                // 存在則不執(zhí)行，但需要將數(shù)據(jù)傳遞下去
                // ack 可以參考 https://www.jianshu.com/p/a5f7fce67803
                ch.ack(msg);
              } else {
                  console.log("Node does not exist.");
                  // 操作完成后則釋放當(dāng)前創(chuàng)建的臨時(shí)節(jié)點(diǎn)
                  client.create(path, null, zookeeper.CreateMode.EPHEMERAL, function (error) {
                    if (error) {
                        console.log("Failed to create node: %s due to: %s.", path, error);
                    } else {
                        console.log("Node: %s is successfully created.", path);
                          
                        // 根據(jù)傳入標(biāo)識(shí)(如書(shū)旗就是37)動(dòng)態(tài)引入js文件(抓書(shū)的操作)
                        moduleCustomize(channel_id).init(channel_book_id)

                        // 傳遞數(shù)據(jù)
                        ch.ack(msg);
                        // 釋放當(dāng)前創(chuàng)建的臨時(shí)節(jié)點(diǎn)
                        client.remove(path, -1, function (error) {
                            if (error) {
                                console.log(error.stack);
                                return;
                            }
                            console.log("Node is deleted.");
                        });

                    }
                  });
                }
              console.log("任務(wù)執(zhí)行完畢")
            });
      }
    });
  }).catch(console.warn);


});
 
client.connect();

命令行啟動(dòng)rab_consumer.js，{} 里面就是我們接收到消息隊(duì)列里面的數(shù)據(jù)了