開發簡單的Kafka應用

　　之前基於集群和單機安裝過kafka，現在利用kafka提供的API構建一個簡單的生產者消費者的項目示例，來跑通kafka的流程，具體過程如下：

　　首先使用eclipse for javaee建立一個maven項目，然后在pom.xml添加如下依賴配置：

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.9.2</artifactId>
        <version>0.8.2.2</version>
    </dependency>

　　這里kafka版本是kafka_2.9.2-0.8.2.2，保存之后maven會自動下載依賴，注意要關閉windows防火牆，盡量專用網絡和外網都要關閉，否則下載的很慢，下載好之后就可以編寫項目代碼了，這里的pom.xml所有配置如下：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>kafkatest</groupId>
  <artifactId>kafkatest</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>kafkatest</name>
  <url>http://maven.apache.org</url>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.9.2</artifactId>
        <version>0.8.2.2</version>
    </dependency>
  </dependencies>
</project>

　　然后，我們建立一個簡單生產者類SimpleProducer，代碼如下：

package test;

import java.util.Properties;

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

public class SimpleProducer {
        private static Producer<Integer,String> producer;
        private final Properties props=new Properties();
        public SimpleProducer(){
                //定義連接的broker list
                props.put("metadata.broker.list", "192.168.1.216:9092");
                //定義序列化類 Java中對象傳輸之前要序列化
                props.put("serializer.class", "kafka.serializer.StringEncoder");
                producer = new Producer<Integer, String>(new ProducerConfig(props));
        }
        public static void main(String[] args) {
                SimpleProducer sp=new SimpleProducer();
                //定義topic
                String topic="mytopic";
                
                //定義要發送給topic的消息
                String messageStr = "This is a message";
                
                //構建消息對象
                KeyedMessage<Integer, String> data = new KeyedMessage<Integer, String>(topic, messageStr);
         
                //推送消息到broker
                producer.send(data);
                producer.close();
        }
}

　　類的代碼很簡單，我這里是kafka單機環境端口就是kafka broker端口9092，這里定義topic為mytopic當然可以自己隨便定義不用考慮服務器是否創建，對於發送消息的話上面代碼是簡單的單條發送，如果發送數據量很大的話send方法多次推送會耗費時間，所以建議把data數據按一定量分組放到List中，最后send一下AarrayList即可，這樣速度會大幅度提高

　　接下來寫一個簡單的消費者類SimpleHLConsumer，代碼如下：

package test;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class SimpleHLConsumer {
        private final ConsumerConnector consumer;
        private final String topic;

        public SimpleHLConsumer(String zookeeper, String groupId, String topic) {
                Properties props = new Properties();
                //定義連接zookeeper信息
                props.put("zookeeper.connect", zookeeper);
                //定義Consumer所有的groupID
                props.put("group.id", groupId);
                props.put("zookeeper.session.timeout.ms", "500");
                props.put("zookeeper.sync.time.ms", "250");
                props.put("auto.commit.interval.ms", "1000");
                consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
                this.topic = topic;
        }

        public void testConsumer() {
                Map<String, Integer> topicCount = new HashMap<String, Integer>();
                //定義訂閱topic數量
                topicCount.put(topic, new Integer(1));
                //返回的是所有topic的Map
                Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer.createMessageStreams(topicCount);
                //取出我們要需要的topic中的消息流
                List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);
                for (final KafkaStream stream : streams) {
                        ConsumerIterator<byte[], byte[]> consumerIte = stream.iterator();
                        while (consumerIte.hasNext())
                                System.out.println("Message from Topic :" + new String(consumerIte.next().message()));
                }
                if (consumer != null)
                        consumer.shutdown();
        }

        public static void main(String[] args) {
                String topic = "mytopic";
                SimpleHLConsumer simpleHLConsumer = new SimpleHLConsumer("192.168.1.216:2181/kafka", "testgroup", topic);
                simpleHLConsumer.testConsumer();
        }

}

　　消費者代碼主要邏輯就是對生產者發送過來的數據做簡單處理和輸出，注意這里的地址是zookeeper的地址並且包括節點/kafka，topic名稱要一致

　　上面兩個類已經可以實現消息的生產和消費了，但是現在服務器需要做一定的配置才可以，否則會拋出異常，就是在之前配置的server.properties基礎之上進行修改，進入kafka安裝目錄下，使用命令 vim config/server.properties 打開配置文件，找到host.name這個配置，首先去掉前面的#注釋，然后把默認的localhost改成IP地址192.168.1.216，因為eclipse遠程運行代碼時讀取到localhost再執行時就是提交到本地了，所以會拋出異常，當然把代碼打成jar包在服務器運行就不會出現這樣的問題了，這里要注意：

　　

　　修改之后保存並退出，然后確保zookeeper的正常運行

　　如果之前kafka正在運行，那么就執行 bin/kafka-server-stop.sh  停止kafka服務，然后再執行

　　 nohup bin/kafka-server-start.sh config/server.properties >> /dev/null & 啟動服務，如果原來就是停止的，那么直接啟動即可

　　啟動之后先運行啟動消費者，消費者處於運行等待

　　

　　然后啟動生產者發送消息，生產者發送完成立即關閉，消費者消費輸出如下：

　　

　　到這里，就完成了kafka從生產到消費簡單示例的開發，消息隊列可以跑通了