Kafka初步学习

/*
* Kafka配置文件, 用于定义producer, consumer
* */
public class KafkaProperties {
 
    //定义端口号
    public static final String ZK = "localhost:2181";
    public static final String TOPIC = "hello_topic";
    public static final String BROKER_LIST = "localhost:9092";
}

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
 
import java.util.Properties;
 
/*
* Kafka生产者
* */
public class KafkaProducer extends Thread{
 
    private String topic;
    //选择kafka.javaapi.producer
    private Producer<Integer, String> producer;
 
    //构造方法，传入topic,生成producer
    public KafkaProducer(String topic) {
 
        this.topic = topic;
 
        //用properties设置ProducerConfig所需要的参数, 这是生成Producer的前提
        //分别是broker_list, 序列化, 握手机制
        Properties properties = new Properties();
        properties.put("metadata.broker.list", KafkaProperties.BROKER_LIST);
        properties.put("serializer.class", "kafka.serializer.StringEncoder");  //此处序列化类用String
        properties.put("request.required.acks", "1");  //可设置为0, 1, -1, 一般生产用1, 最严谨是-1, 不能用0
 
        producer = new Producer<Integer, String>(new ProducerConfig(properties));
    }
 
    //用线程来启动producer
    @Override
    public void run() {
 
        int messageNo = 1;
 
        while(true) {
            String message = "massage_" + messageNo;
            producer.send(new KeyedMessage<Integer, String>(topic, message));
            System.out.println("send: " + message);
 
            messageNo++;
 
            //2s间隔发送一次
            try {
                Thread.sleep(2000);
            } catch(Exception e) {
                e.printStackTrace();
            }
        }
    }
}

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
 
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
 
import static com.sun.org.apache.xml.internal.security.keys.keyresolver.KeyResolver.iterator;
import static javafx.scene.input.KeyCode.V;
 
/*
* Kafka消费者
* */
public class KafkaConsumer extends Thread {
 
    private String topic;
 
    public KafkaConsumer(String topic) {
 
        this.topic = topic;
    }
 
    //ConsumerConnector选择kafka.javaapi.consumer包
    //此处是要创建consumer连接器, 而不是创建consumer, 区别于producer
    private ConsumerConnector createConnector() {
 
        //同样地设置ConsumerConfig对象的属性
        //需要设置ZK
        Properties properties = new Properties();
        properties.put("zookeeper.connect", KafkaProperties.ZK);
        properties.put("group.id", KafkaProperties.GROUP_ID);
        return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    }
 
 
    //线程启动consumer
    @Override
    public void run() {
 
        ConsumerConnector consumer = createConnector();
 
        //由于createMessageStreams需要传入一个Map, 所以创建一个
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        //map中放入topic和kafka stream的数量
        topicCountMap.put(topic, 1);
 
        //创建messageStream, 从源码中可看出它的数据类型
        //String是topic, List是数据比特流
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStream = consumer.createMessageStreams(topicCountMap);
        //获取每次的数据
        KafkaStream<byte[], byte[]> byteStream = messageStream.get(topic).get(0);
 
        //数据流进行迭代
        ConsumerIterator<byte[], byte[]> iterator = byteStream.iterator();
 
        while (iterator.hasNext()) {
 
            //由于iterator里面的是byte类型，要转为String
            String message = new String(iterator.next().message());
            System.out.println("receive:" + message);
        }
    }
}