import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
private void configureKafkaBolt(TopologyBuilder builder, Config config) {
	String topic = topologyConfig.getProperty("kafka.topic");
	Properties props = new Properties();
	props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
	props.put(ProducerConfig.CLIENT_ID_CONFIG, "storm-kafka-producer");
	props.put("metadata.broker.list", brokerUrl);
	props.put("serializer.class", "kafka.serializer.StringEncoder");
	props.put("request.required.acks", "1");
	config.setMaxSpoutPending(20);
	config.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
	KafkaBolt<String, String> kafkaBolt = new KafkaBolt<String, String>().withTopicSelector(new DefaultTopicSelector(topic))
									.withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper<String, String>("key", "log"));
	builder.setBolt("KafkaBolt", kafkaBolt, 3).shuffleGrouping(SPOUT_ID).setDebug(DEBUG);
}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
/**
 * A topology that produces random sentences using {@link RandomSentenceSpout} and
 * publishes the sentences using a KafkaBolt to kafka "test" topic.
 *
 * @return the storm topology
 */
public StormTopology buildProducerTopology() {
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("spout", new RandomSentenceSpout(), 2);
    /**
     * The output field of the RandomSentenceSpout ("word") is provided as the boltMessageField
     * so that this gets written out as the message in the kafka topic.
     */
    KafkaBolt bolt = new KafkaBolt()
            .withTopicSelector(new DefaultTopicSelector(KAFKA_TOPIC))
            .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper("key", "word"));
    builder.setBolt("forwardToKafka", bolt, 1).shuffleGrouping("spout");
    return builder.createTopology();
}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
/**
     * Returns the storm config for the topology that publishes sentences to kafka "test" topic using a kafka bolt.
     * The KAFKA_BROKER_PROPERTIES is needed for the KafkaBolt.
     *
     * @return the topology config
     */
    public Config getProducerConfig() {
        Config conf = new Config();
        conf.setMaxSpoutPending(20);
        Properties props = new Properties();
        props.put("metadata.broker.list", brokerUrl);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "storm-kafka-producer");
        conf.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
        return conf;
    }
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
public static TridentKafkaStateFactory getTridentKafkaStateFactory(String topicName, String brokerList, String keyField, String messageField, Map topologyConfig) {
    Properties props = new Properties();
    props.put("metadata.broker.list", brokerList);
    props.put("request.required.acks", "1");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    topologyConfig.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);

    return new TridentKafkaStateFactory()
            .withKafkaTopicSelector(new DefaultTopicSelector(topicName))
            .withTridentTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper(keyField, messageField));
}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
public static KafkaBolt<String, String> getKafkaBolt(String topicName, String brokerList, String keyField,
                                                     String messageField, Map topologyConfig) {
    Properties props = new Properties();
    props.put("metadata.broker.list", brokerList);
    props.put("request.required.acks", "1");
    props.put("serializer.class", "kafka.serializer.StringEncoder");
    topologyConfig.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, props);
    return new KafkaBolt<String, String>()
            .withTopicSelector(new DefaultTopicSelector(topicName))
            .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper<String, String>(keyField, messageField));
}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
/**
 * 
 * This example is very dangerous to the consistency of your bank accounts. Guess why, or read the
 * tutorial.
 * 
 * @throws AlreadyAliveException
 * @throws InvalidTopologyException
 * @throws AuthorizationException
 */
public static void main(String... args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException {

	// starting to build topology
	TopologyBuilder builder = new TopologyBuilder();

	// Kafka as a spout
	builder.setSpout(IDs.kafkaSpout, new KafkaSpoutBuilder(Conf.zookeeper, Conf.inputTopic).build());

	// bolt to println data
	builder.setBolt(IDs.printlnBolt, new PrintlnBolt()).shuffleGrouping(IDs.kafkaSpout);

	// bolt to perform transactions and simulate bank accounts
	builder.setBolt(IDs.userAccountBolt, new BankAccountBolt()).shuffleGrouping(IDs.kafkaSpout);

	// Kafka as a bolt -- sending messages to the output topic
	KafkaBolt<Object, Object> bolt = new KafkaBolt<>().withTopicSelector(new DefaultTopicSelector(Conf.outputTopic))
			.withTupleToKafkaMapper(new TransactionTupleToKafkaMapper());
	builder.setBolt(IDs.kafkaBolt, bolt).shuffleGrouping(IDs.userAccountBolt);

	// submit topolody to local cluster
	new LocalCluster().submitTopology(IDs.kafkaAccountsTopology, topologyConfig(), builder.createTopology());

	// wait a while, then simulate random transaction stream to Kafka
	Sleep.seconds(5);
	KafkaProduceExample.start(2000);

}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
private Map getKafkaConfig(Map options) {
  Map kafkaConfig = new HashMap();
  Map brokerConfig = new HashMap();
  String brokers = (String) Utils.get(options, BROKER_LIST, "localhost:9092");
  String topic = (String) Utils.get(options, TOPIC, KafkaUtils.DEFAULT_TOPIC);
  brokerConfig.put("metadata.broker.list", brokers);
  brokerConfig.put("serializer.class", "kafka.serializer.StringEncoder");
  brokerConfig.put("key.serializer.class", "kafka.serializer.StringEncoder");
  brokerConfig.put("request.required.acks", "1");
  kafkaConfig.put(KafkaBolt.KAFKA_BROKER_PROPERTIES, brokerConfig);
  kafkaConfig.put(KafkaBolt.TOPIC, topic);
  return kafkaConfig;
}
 

import storm.kafka.bolt.KafkaBolt; //导入依赖的package包/类
private boolean initializeKafkaBolt(String name) {
	try {

		String messageUpstreamComponent = messageComponents
				.get(messageComponents.size() - 1);

		System.out.println("[OpenSOC] ------" + name
				+ " is initializing from " + messageUpstreamComponent);

		Map<String, String> kafka_broker_properties = new HashMap<String, String>();
		kafka_broker_properties.put("zk.connect",
				config.getString("kafka.zk"));
		kafka_broker_properties.put("metadata.broker.list",
				config.getString("kafka.br"));

		kafka_broker_properties.put("serializer.class",
				"com.opensoc.json.serialization.JSONKafkaSerializer");

		kafka_broker_properties.put("key.serializer.class",
				"kafka.serializer.StringEncoder");

		String output_topic = config.getString("bolt.kafka.topic");

		conf.put("kafka.broker.properties", kafka_broker_properties);
		conf.put("topic", output_topic);

		builder.setBolt(name, new KafkaBolt<String, JSONObject>(),
				config.getInt("bolt.kafka.parallelism.hint"))
				.shuffleGrouping(messageUpstreamComponent, "message")
				.setNumTasks(config.getInt("bolt.kafka.num.tasks"));
	} catch (Exception e) {
		e.printStackTrace();
		System.exit(0);
	}
	return true;
}