<!-- java程式的Kakfa-->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.10</artifactId>
      <version>0.10.0.0</version>
    </dependency>

}

生產者：

public class KafKaProducer {
    private final Producer<String,String> producer;
    public final static String TOPIC="linlin";

    private KafKaProducer(){
        Properties properties=new Properties();
        //此處設定kafka的埠
        properties.put("metadata.broker.list","127.0.0.1:9092");
       // properties.put("zk.connect","127.0.0.1:2181");
        //配置value的序列化類
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        properties.put("key.serializer.class","kafka.serializer.StringEncoder");
        properties.put("request.required.acks", "-1");

        producer=new Producer<String, String>(new ProducerConfig(properties));
    }


    void produce(){
        int messageNo=1000;
        final int count=10000;
        while (messageNo<count){
            String key=String.valueOf(messageNo);
            String data ="hello kafka message"+key;
            producer.send(new KeyedMessage<String, String>(TOPIC,key,data));
            System.out.println("傳送訊息："+data);
            messageNo++;
        }


    }

  public static void main(String[] args){
       new KafKaProducer().produce();
  }

}