Apache Kafka 初步

lrx182125

1:
 
下载ApacheKafka,
安装:
tar -xvf kafka.tgz
 
2:
修改配置文件:
 
(1):
kafka/conf/zookeeper.properties
中

dataDir=/disks/sdb3/soft/Kafka/Data/ZooKeeper/2.9.2/dataDir
 
(2):
kafka/conf/server.properties
中

log.dirs=/disks/sdb3/soft/Kafka/Data/KafkaData/2.9.2/logs
host.name=192.168.56.3
 
此处的host.name为本机IP(重要),如果不改,则客户端会抛出:Producer connection to localhost:9092 unsuccessful 错误!
 
3:
启动:
 
启动zookeeper:

bin/zookeeper-server-start.sh config/zookeeper.properties
 
启动kafka:

bin/kafka-server-start.sh config/server.properties
 
4:
Java客户端:

package com.test.search.test.kafka;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.junit.Test;
import com.google.common.collect.Maps;
public class KafkaTest {
private static final String KAFKA_BROKER_LIST = "192.168.56.3:9092";
private static final String KAFKA_ZOOKEEPER_LIST = "192.168.56.3:2181";
private static final String KAFKA_TOPIC = "CustomTopic";
class Producer extends Thread {
private final kafka.javaapi.producer.Producer<Integer, String> producer;
private final String topic;
private final Properties props = new Properties();
public Producer(String topic) {
props.put("serializer.class", "kafka.serializer.StringEncoder");
// props.put("metadata.broker.list", "localhost:9092");
props.put("metadata.broker.list", KAFKA_BROKER_LIST);
props.put("request.required.acks", "1");
// Use random partitioner. Don't need the key type. Just set it to Integer.
// The message is of type String.
producer = new kafka.javaapi.producer.Producer<Integer, String>(new ProducerConfig(props));
this.topic = topic;
}
public void sendMsg(String msg) {
producer.send(new KeyedMessage<Integer, String>(topic, msg));
System.out.println("已发送: " + msg);
}
public void run() {
int messageNo = 1;
while (true) {
String messageStr = new String("Message_" + messageNo);
producer.send(new KeyedMessage<Integer, String>(topic, messageStr));
messageNo++;
}
}
}
private static ConsumerConfig createConsumerConfig() {
Properties props = new Properties();
props.put("zookeeper.connect", KAFKA_ZOOKEEPER_LIST);
props.put("group.id", "0");
props.put("zookeeper.session.timeout.ms", "400");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
return new ConsumerConfig(props);
}
class Consumer extends Thread {
private final ConsumerConnector consumer;
private final String topic;
public Consumer(String topic) {
consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());
this.topic = topic;
}
public void receiveMsg() {
Map<String, Integer> topicCountMap = Maps.newHashMap();
topicCountMap.put(topic, new Integer(1));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext()) {
System.out.println(new String(it.next().message()));
}
}
public void run() {
Map<String, Integer> topicCountMap = Maps.newHashMap();
topicCountMap.put(topic, new Integer(1));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0);
ConsumerIterator<byte[], byte[]> it = stream.iterator();
while (it.hasNext())
System.out.println(new String(it.next().message()));
}
}
@Test
public void testSendMsg() {
Producer producer = new Producer(KAFKA_TOPIC);
producer.sendMsg("{'cardid':21,'opt':'c'}");
}
@Test
public void testReceiveMsg() {
Consumer consumer = new Consumer(KAFKA_TOPIC);
consumer.receiveMsg();
}
}

 
 5:
　　参考:
　　http://kafka.apache.org/documentation.html 
　　http://stackoverflow.com/questions/17808988/using-kafka-java-producer-send-a-message-producer-connection-to-localhost9092