怎么用Storm IPResolutionBolt寫爬蟲

本篇內容介紹了“怎么用Storm IPResolutionBolt寫爬蟲”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

package com.digitalpebble.storm.crawler.bolt;

import java.net.InetAddress;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class IPResolutionBolt extends BaseRichBolt {

	public static final Logger LOG = LoggerFactory
			.getLogger(IPResolutionBolt.class);

	OutputCollector _collector;

	@SuppressWarnings("unchecked")
	public void execute(Tuple tuple) {
		String url = tuple.getStringByField("url");
		HashMap<String, String[]> metadata = null;

		
		//在這里判斷我們的tuple是否包含著Meradata
		if (tuple.contains("metadata"))
			metadata = (HashMap<String, String[]>) tuple
					.getValueByField("metadata");
					
	        //這里的Metadata是一個HashMap，持有的是一個 <String,String[]> 的對象組合

		String ip = null;
		String host = "";

		URL u;
		try {
			u = new URL(url);
			host = u.getHost();
		} catch (MalformedURLException e1) {
			LOG.warn("Invalid URL: " + url);
			// ack it so that it doesn't get replayed
			_collector.ack(tuple);
			return;
		}

		try {
			long start = System.currentTimeMillis();
			final InetAddress addr = InetAddress.getByName(host);
			ip = addr.getHostAddress();
			long end = System.currentTimeMillis();

			LOG.info("IP for: " + host + " > " + ip + " in " + (end - start)
					+ " msec");

					
		        //在這里我們發(fā)射  url，ip，metadata 并且針對tuple做一個Ack
			_collector.emit(tuple, new Values(url, ip, metadata));
			_collector.ack(tuple);
		} catch (final Exception e) {
			LOG.warn("Unable to resolve IP for: " + host);
			_collector.fail(tuple);
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("url", "ip", "metadata"));
	}

	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		_collector = collector;
	}

}

          在這里我們需要關注， declareOutputFields在設定我們的Tuple records對象的時候，是傳遞的“url”，“ip”，“metadata”， 而不是一個封裝好的對象。 

        一旦我們傳遞的records的數(shù)量比較多。那么請寧務必將傳遞的值設置為對象。并且在接受方，getValues（0）的方式取得。

“怎么用Storm IPResolutionBolt寫爬蟲”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！