【准备】
1 : zookeeper集群 搭建在224, 225, 226
2 : kafka使用3个节点224, 225, 226
修改配置文件config/server.properties
broker.id=224
host.name=192.168.1.224
log.dirs=/usr/local/kafka_2.10-0.8.2.0/kafka_log
复制到其他两个节点，然后修改对应节点上的config/server.pro
3 : 分别启动三个节点
[root@hadoop2004 bin]# ./kafka-server-start.sh -daemon ../config/server.properties
4 : 创建一个“test”主题
[root@hadoop2004 bin]# ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic test
5 : 查看详细信息
[root@hadoop2004 bin]# ./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
6 : 去zookeeper上查看kafka集群
[zk: localhost:2181(CONNECTED) 0] ls /
[hbase, hadoop-ha, admin, zookeeper, consumers, config, controller, brokers, controller_epoch]
[zk: localhost:2181(CONNECTED) 1] ls /brokers
[topics, ids]
[zk: localhost:2181(CONNECTED) 2] ls /brokers/ids
[224, 225, 226]
[zk: localhost:2181(CONNECTED) 3] ls /brokers/topics
[test, lvfang]
[zk: localhost:2181(CONNECTED) 4] ls /brokers/topics/test
[partitions]
[zk: localhost:2181(CONNECTED) 5] ls /brokers/topics/test/partitions
[2, 1, 0]
[zk: localhost:2181(CONNECTED) 6]
【编写java代码】
创建java工程，导入jar包（jar来自tar的lib下）
【1 ：生产者类】KafkaProduce
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;
/**
* kafka生产者类
* @author lvfang
*
*/
public class KafkaProduce extends Thread {
private String topic;//主题
public KafkaProduce(String topic){
super();
this.topic = topic;
}
//创建生产者
private Producer createProducer(){
Properties properties = new Properties();
//声明zookeeper集群
properties.put("zookeeper.connect",
"192.168.1.224:2181," +
"192.168.1.225:2181," +
"192.168.1.226:2181");
properties.put("serializer.class", StringEncoder.class.getName());
// 声明kafka集群的 broker
properties.put("metadata.broker.list",
"192.168.1.224:19092," +
"192.168.1.225:19092," +
"192.168.1.226:19092");
return new Producer<Integer, String>(new ProducerConfig(properties));
}
@Override
public void run() {
//创建生产者
Producer producer = createProducer();
int i=0;
//循环发送消息到kafka
while(true){
producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));
try {
//发送消息的时间间隔
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
// 使用kafka集群中创建好的主题 test
new KafkaProduce("test").start();
}
}
【2 ： kafka集群中消费主题test的数据】
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginnin
【3 ： 开启生产者的java线程】
这时你会发现接收者已经可以接受到消息了
【4 ： 消费者类】KafkaCusumer
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;
/**
* kafka消费者类
* @author lvfang
*
*/
public class KafkaCusumer extends Thread {
private String topic;//主题
public KafkaCusumer(String topic){
super();
this.topic = topic;
}
//创建消费者
private ConsumerConnector createConsumer(){
Properties properties = new Properties();
//声明zookeeper集群链接地址
properties.put("zookeeper.connect",
"192.168.1.224:2181," +
"192.168.1.225:2181," +
"192.168.1.226:2181");
//必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据
//当前消费者的组名称
properties.put("group.id", "group1");
return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
}
@Override
public void run() {
//创建消费者
ConsumerConnector consumer = createConsumer();
//主题数map
Map<String, Integer> topicCountMap = new HashMap<>();
// 一次从topic主题中获取一个数据
topicCountMap.put(topic, 1);
//创建一个获取消息的消息流
Map<String,List<KafkaStream<byte[], byte[]>>> messageStreams = consumer.createMessageStreams(topicCountMap);
// 获取每次接收topic主题到的这个数据
KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);
ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
//循环打印
while (iterator.hasNext()) {
String message = new String(iterator.next().message());
System.out.println("接收到: " + message);
}
}
public static void main(String[] args) {
// 使用kafka集群中创建好的主题 test
new KafkaCusumer("test").start();
}
}
开启消费者线程你会发现消费者也可以接收到并消费
【流程】
生产者生产数据 ----> 推送 ----> kafka ----> 获取 ----> 消费者获取数据
【补充说明】
补充说明：
1、public Map<String, List<KafkaStream<byte[], byte[]>>> createMessageStreams(Map<String, Integer> topicCountMap)，其中该方法的参数Map的key为topic名称，value为topic对应的分区数，譬如说如果在kafka中不存在相应的topic时，则会创建一个topic，分区数为value，如果存在的话，该处的value则不起什么作用
2、关于生产者向指定的分区发送数据，通过设置partitioner.class的属性来指定向那个分区发送数据，如果自己指定必须编写相应的程序，默认是kafka.producer.DefaultPartitioner,分区程序是基于散列的键。
3、在多个消费者读取同一个topic的数据，为了保证每个消费者读取数据的唯一性，必须将这些消费者group_id定义为同一个值，这样就构建了一个类似队列的数据结构，如果定义不同，则类似一种广播结构的。
4、在consumerapi中，参数设计到数字部分，类似Map<String,Integer>,
numStream,指的都是在topic不存在的时，会创建一个topic，并且分区个数为Integer,numStream,注意如果数字大于broker的配置中num.partitions属性，会以num.partitions为依据创建分区个数的。
5、producerapi，调用send时，如果不存在topic，也会创建topic，在该方法中没有提供分区个数的参数，在这里分区个数是由服务端broker的配置中num.partitions属性决定的
