kafka java demo

环境：ubuntu10 2台（32位）+JDK1.8（32位）+kafka2.11+Intellij15

 

目标：Java启动一个Producer,启动一个Consumer,Linux启动一个Consumer.

观察3者是否能相互通信。

 

注意到，Java的Producer和Consumer全是用maven构建的，父项目是kafka_demo,他们两个是module.

 

1、Java Producer Demo:

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

import java.util.Properties;

/**
 * Created by Germmy on 2016/7/10.
 */
public class KafkaProducer {

    private  final Producer<String,String> producer;

    public final static String TOPIC="TEST-TOPIC";

    private  KafkaProducer(){
        Properties props=new Properties();
        props.put("metadata.broker.list","192.168.200.129:9092");
        props.put("serializer.class","kafka.serializer.StringEncoder");
        props.put("key.serializer.class","kafka.serializer.StringEncoder");
        props.put("request.required.acks","-1");
        producer=new Producer<String, String>(new ProducerConfig(props)) ;
    }


    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();
    }

}

 

2、Java Consumer Demo:

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

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

/**
 * Created by Germmy on 2016/7/10.
 */
public class KafkaConsumer {


    private  final ConsumerConnector consumer;

    public final static String TOPIC="TEST-TOPIC";

    private KafkaConsumer(){
        Properties props=new Properties();
        props.put("zookeeper.connect","192.168.200.129:2181");
        props.put("group.id","jd-group");//消费组是什么概念？

        props.put("zookeeper.session.timeout.ms","60000");
        props.put("zookeeper.sync.time.ms","200");
        props.put("auto.commit.interval.ms","1000");
        props.put("auto.offset.reset","smallest");

        props.put("serializer.class","kafka.serializer.StringEncoder");

        ConsumerConfig config=new ConsumerConfig(props);

        consumer=kafka.consumer.Consumer.createJavaConsumerConnector(config);
    }


    void consume(){
        Map<String,Integer> topicCountMap=new HashMap<String, Integer>();
        topicCountMap.put(KafkaConsumer.TOPIC,new Integer(1));
        StringDecoder keyDecoder=new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder=new StringDecoder(new VerifiableProperties());

        Map<String,List<KafkaStream<String,String>>> consumerMap=
                consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);

        KafkaStream<String,String> stream=consumerMap.get(KafkaConsumer.TOPIC).get(0);
        ConsumerIterator<String,String> it=stream.iterator();
        while(it.hasNext()){
            System.out.println(it.next().message());
        }
    }


    public  static  void main(String[] args){
        new KafkaConsumer().consume();
    }

}

 

注意到，Consumer连接zookeeper的超时时间需要设置长一点，之前的版本是4秒，会报连接超时异常。我这里设置的是60S，参考链接

 

 

3、父POM：

<?xml version="1.0" encoding="UTF-8"?>
<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>com.germmy</groupId>
    <artifactId>kafkademo</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>kafka_producer</module>
        <module>kafka_consumer</module>
    </modules>

    <dependencies>

        <!--<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>0.10.0.0</version>
        </dependency>-->

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.10</artifactId>
            <version>0.8.0</version>
            <exclusions>
                <exclusion>
                    <groupId>javax.jms</groupId>
                    <artifactId>jms</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jmx</groupId>
                    <artifactId>jmxri</artifactId>
                </exclusion>
                <exclusion>
                    <groupId>com.sun.jdmk</groupId>
                    <artifactId>jmxtools</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

    </dependencies>


</project>

 

 

注意到，这里引入的client客户端要去除对jms的依赖，否则会报如下错：

Could not transfer artifact com.sun.jdmk:jmxtools:jar:1.2.1 from/to java.net (https://maven-repository.dev.java.net/nonav/repository): No connector available to access repository java.net (https://maven-repository.dev.java.net/nonav/repository) of type legacy using the available factories 

overstack上的解决的方案

 

 

至此，demo已经成功运行。

 

4、遗留的问题如下：

      * Producer用mvn打包失败，具体原因待查。

      * props.put("group.id","jd-group");//消费组的概念待明确。

      * intellij启动多个main后，console是直接TAB在一起的，根本不要显式切换。

      * alt+1，直接打开projectViews.

      * 点击左下角的正方形，可以打开或者关闭所有的侧边栏。

      * 关于用0.8的client报jms的错误问题，有人说是因为它去maven2的仓库中了，所以要将repositories设置指向maven3的仓库。参考链接。

      * 用intellij有时maven不会自动下载依赖，此时可以用cmd直接敲mvn compile命令试试。

      * 在git命令中已经和远程仓库关联，但是在intellij中还不知道如何关联。

 

 

 

-------------------------------------------------------------------------------------------------

本文参考的java demo