kafka简介
高流量的活跃数据是无法确定其大小的，因为他可能随时的变化，比如商家可能促销，节假日打折，突然又冒出一个跳楼价等等。所有的数据可能是数量级的往上递增。 传统日志分析方式都是需要离线，而且操作起来比较复杂，根本无法满足实时的分析。另一方面，现有的消息队列系统只能达到近似实时的分析，因为无法消费大量的持久化在队列系统上的信息。Kafka的目标就是能够成为一个高效的队列平台，无论是处理离线的信息还是在线的信息。
安装伪分布式kafka
cd /usr/local
tar -zxvf kafka_2.10-0.8.2.0.tgz
mv kafka_2.10-0.8.2.0 kafka
cd /usr/local/kafka/
启动Kafka自带的ZooKeeper，后台运行
bin/zookeeper-server-start.sh config/zookeeper.properties >/dev/null 2>&1 &
启动Kafka服务，后台运行
bin/kafka-server-start.sh config/server.properties >/dev/null 2>&1 &
创建一个Kafka的主题，连接到本地zk，副本因子1，分区1，主题名是test
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
查看ZooKeeper上Kafka的主题
bin/kafka-topics.sh --list --zookeeper localhost:2181
查看Kafka的主题详情
bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
创建生产者
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
创建消费者
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
安装完全分布式kafka，在h5 h6 h7节点上
在h5节点上安装Kafka，
要求启动ZooKeeper集群。
cd /usr/local
tar -zxvf kafka_2.10-0.8.2.0.tgz
mv kafka_2.10-0.8.2.0 kafka
cd /usr/local/kafka/
vi config/server.properties
broker.id=36 ##必须是数字
host.name=h6 ##可以是IP、主机名、域名
log.dirs=/usr/local/kafka/logs
scp -rq /usr/local/kafka/ h6:/usr/local
scp -rq /usr/local/kafka/ h7:/usr/local
创建一个Kafka的主题，连接到zk集群，副本因子3，分区3，主题名是test
bin/kafka-topics.sh --create --zookeeper h5:2181 --topic test111 --replication-factor 3 --partitions 3
查看Kafka的主题详情
bin/kafka-topics.sh --describe --zookeeper h5:2181 --topic test111
zkCli.sh
ls /brokers/topics/test/
使用java代码实现kafka的生产者和消费者
1、生产者
package com.mengyao.kafka;
import java.util.Properties;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
public class KafkaProducerTest extends Thread {
private String topic;
public KafkaProducerTest(){
}
public KafkaProducerTest(String topic){
this.topic = topic;
}
private Producer<Integer, String> getProducer(Properties prop) {
return new Producer<Integer, String>(new ProducerConfig(prop));
}
private Properties getProperties() {
Properties prop = new Properties();
prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
prop.put("serializer.class", StringEncoder.class.getName());
prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
return prop;
}
@Override
public void run() {
Properties prop = getProperties();
Producer<Integer, String> producer = getProducer(prop);
int i = 0;
while (true) {
producer.send(new KeyedMessage<Integer, String>(topic, "msg:"+i++));
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
new KafkaProducerTest("test111").start();
}
}
2、消费者
package com.mengyao.kafka;
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;
import kafka.serializer.StringEncoder;
public class KafkaConsumerTest extends Thread {
private String topic;
public KafkaConsumerTest() {
}
public KafkaConsumerTest(String topic) {
this.topic = topic;
}
private ConsumerConnector getConsumer(Properties prop) {
return Consumer.createJavaConsumerConnector(new ConsumerConfig(prop));
}
private Properties getProperties() {
Properties prop = new Properties();
prop.put("zookeeper.connect", "h5:2181,h6:2181,h7:2181");
prop.put("serializer.class", StringEncoder.class.getName());
prop.put("metadata.broker.list", "h5:9092,h6:9092,h7:9092");
prop.put("group.id", "group1");
return prop;
}
@Override
public void run() {
Properties prop = getProperties();
ConsumerConnector consumer = getConsumer(prop);
HashMap<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, 1);
Map<String, List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]> kafkaStream = messageStreams.get(topic).get(0);
ConsumerIterator<byte[], byte[]> iterator = kafkaStream.iterator();
while (iterator.hasNext()) {
final String msg = new String(iterator.next().message());
System.out.println(msg);
}
}
public static void main(String[] args) {
new KafkaConsumerTest("test111").start();
}
}