storm集成kafka

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;

import org.springframework.beans.factory.annotation.Value;

public class KafkaUtil {
    
    @Value("#{sys[‘connect‘]}")
    private static String zkConnect ;
    @Value("#{sys[‘metadata.broker.list‘]}")
    private static String brokerList;
    @Value("#{sys[‘request.required.acks‘]}")
    private static String ack;
    
    private static Producer<String, String> producer = null;
    
    /*static{
        Properties p = PropertiesUtil.getProperties("kafka.properties");
        zkConnect = (String) p.get("zk.connect");
        brokerList = (String) p.get("metadata.broker.list");
        ack = (String) p.get("request.required.acks");
        topic = (String) p.get("topic.imeidata");
    }
    */
    public static Producer<String,String> getProducer(){
        if(producer == null){
            Properties p = PropertiesUtil.getProperties("kafka.properties");
            zkConnect = (String) p.get("zk.connect");
            brokerList = (String) p.get("metadata.broker.list");
            ack = (String) p.get("request.required.acks");
            
            Properties props = new Properties();
            props.put("zk.connect", zkConnect);
            props.put("metadata.broker.list", brokerList);
            props.put("serializer.class", "kafka.serializer.StringEncoder");
            props.put("request.required.acks", ack);
            props.put("producer.type", "async");//是否同步 sync：同步   async：异步  
            props.put("partitioner.class", "com.kafka.SendPartitioner");//发送到多个分区进行分布式存储的分区算法类
            
            props.put("request.timeout.ms", "50000");
            props.put("queue.buffering.max.ms", "10000");//默认值5000  异步模式下，每隔此时间间隔会将缓冲的消息提交一次
            props.put("batch.num.messages", "1000");//默认值200  异步模式下，一次批量提交消息的条数，
                                                    //但如果间隔时间超过 queue.buffering.max.ms 的值，不管有没有达到批量提交的设值，都会进行一次提交
            ProducerConfig config = new ProducerConfig(props);
            producer = new Producer<String, String>(config);
        }
        return producer;
    }
   
}

import kafka.producer.Partitioner;
import kafka.utils.VerifiableProperties;

public class SendPartitioner implements Partitioner{

    public SendPartitioner(VerifiableProperties verifiableProperties) {}

    @Override
    public int partition(Object key, int numPartitions) {
        try {
            return Math.abs(key.hashCode() % numPartitions);
        } catch (Exception e) {
            return Math.abs(key.hashCode() % numPartitions);
        }
    }

}

import com.util.PropertiesUtil;

import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
    
public class Topology {
    
    private static String topic;
    private static String brokerZkStr;
    private static String brokerZkPath;
    private static String offset;
    private static String app;
    
    static{
        Properties p = PropertiesUtil.getProperties("storm.properties");
        brokerZkStr = (String) p.get("brokerZkStr");
        brokerZkPath = (String) p.get("brokerZkPath");
        topic = (String) p.get("kafka.topic");
        offset = (String) p.get("kafka.offset");
        app = (String) p.get("kafka.app");
         
    }
    
    
    public static void main(String[] args) throws InterruptedException {  
        ZkHosts zk = new ZkHosts(brokerZkStr,brokerZkPath);
        SpoutConfig spoutConf = new SpoutConfig(zk, topic, 
                offset,//偏移量 offset 的根目录
                app);//对应一个应用
        
        List<String> zkServices = new ArrayList<>();
        
        for(String str : zk.brokerZkStr.split(",")){
            zkServices.add(str.split(":")[0]);
        }
        
        spoutConf.zkServers = zkServices;
        spoutConf.zkPort = 2181;
        spoutConf.forceFromStart = false;// true:从头消费  false:从offset处消费
        spoutConf.socketTimeoutMs = 60 * 1000;
        spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
        
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("spout", new KafkaSpout(spoutConf),4);
        //builder.setSpout("spout", new TestSpout(),5);
        builder.setBolt("bolt1", new GetMsgBolt(),4).shuffleGrouping("spout");
        
        Config config = new Config();
        config.setDebug(false);
        config.setNumWorkers(4);
        if(args.length>0){
            try {
                StormSubmitter.submitTopology(args[0], config, builder.createTopology());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("MyTopology", config, builder.createTopology());
        }
    }  
}