kafka快速上手
安装(以windows为例)
安装非常简单,从这里下载,下载完成后解压到一个目录就好了.
简单使用
首先使用kafka的一个流程就是生产者生产消息,发送给kafka集群,然后消费者从kafka集群中获取消息进行消费.
要启动
kafka需要先启动zookeeper,因为ZooKeeper是通过冗余服务实现高可用性的,也就是说在分布式环境中,如何保证kafka集群的高可用.zookeeper会来做leader选取,当消费者准备发消息时,会从zookeeper中获取一个可用的消息服务器地址,然后连接进行发送,保证党集群内有服务器宕机并不影响整体的使用.
开发实例入门教程-kafka 快速入门">
来自slideshare的一张图
1.启动自带的简易zookeeper.
进行解压目录的bin/windows目录
zookeeper-server-start.bat ../../config/zookeeper.properties
执行命令启动,从zookeeper.properties中会看到.zookeeper会开发一个clientPort=2181,2181的端口给消费者使用,其实也可以给生产者使用,但是在0.8.0版本后，producer不再通过zookeeper连接broker,
而是通过brokerlist(192.168.0.1:9092,192.168.0.2:9092,192.168.0.3:9092配置,直接和
broker连接，只要能和一个broker连接上就能够获取到集群中其他broker上的信息,绕过了zookeeper.
2.启动kafka服务
kafka-server-start.bat ../../config/server.properties
执行启动,另一个命令行窗口,同样的.查看配置问题,会知道kafka的服务会在port=9092 ,9092端口打开.
3.注册一个topic
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1
--partitions 1 --topic test
这个命令中,create表示创建.zookeeper
和后面的地址表示kafka使用本机2181端口开放的zookeeper保持高可用.replication-factor表示消息只冗余一份,目前我们只有一个kafka机器,broker,partitions
表示一份分区,分区是kafka的另一个概念,大致是说,同一topic内部的消息按照一定的key和算法被分区(partition)存储在不同的位置上，这个下次写好了.这样已经在kafka注册了一个名为test的消息topic了.
4.使用简易的控制台生产者模拟
kafka-console-producer.bat --broker-list localhost:9092 --topic test
前面说过了.新版本生产者直接通过brokerlist来连接kafka,目前只有一台,所以就一个地址,准备向test这个topic发送消息.
5.使用简易的控制台消费者模拟
kafka-console-consumer.bat --zookeeper localhost:2181 --topic test
这个前面也说过了.消费者使用zookeeper获取可用的broker列表,然后拉去消息,并且还有一些offset同步的问题.和分区,文件存储一起的一个概念,下次写.
6.开始生产和消费消息
至此,已经开了四个控制台窗口了..在producer窗口里,随便打几个字,然后enter,在消费者的窗口里将会显示出来.
实际测试图
其他问题
实际可能不那么顺利,如果你启动kafka或者其他应用的时候,有错误提示,提示无法创建虚拟机vm这样的.那么修改一下对应的bat脚本.就好了
启动错误,vm的heap申请是1G,如果你机器内存不够,改成512M,或者更小的就好了.
kafka快速开发实例
1 新建一个maven工程,引入依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>0.8.2.1</version>
</dependency>
2 编写配置文件
public interface KafkaProperties {
public final static String ZK = "127.0.0.1:2181";
public final static String GROUP_ID = "test_group1";
public final static String TOPIC = "test";
public final static String BROKER_LIST = "127.0.0.1:9092";
public final static String SESSION_TIMEOUT = "20000";
public final static String SYNC_TIMEOUT = "20000";
public final static String INTERVAL = "1000";
}
3 编写生产者
public class KafkaProducer extends Thread {
private Producer<Integer, String> producer;
private String topic;
private Properties props = new Properties();
private final int SLEEP = 1000 * 3;
public KafkaProducer(String topic) {
props.put("serializer.class", "kafka.serializer.StringEncoder");
//生产者直接和broker列表连接
props.put("metadata.broker.list", KafkaProperties.BROKER_LIST);
producer = new Producer<Integer, String>(new ProducerConfig(props));
this.topic = topic;
}
@Override
public void run() {
int offsetNo = 1;
while (true) {
String msg = new String("Message_" + offsetNo);
System.out.println("Send->[" + msg + "]");
producer.send(new KeyedMessage<Integer, String>(topic, msg));
offsetNo++;
try {
sleep(SLEEP);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
4 编写消费者
public class KafkaConsumer extends Thread {
private ConsumerConnector consumer;
private String topic;
private final int SLEEP = 1000 * 3;
public KafkaConsumer(String topic) {
consumer = Consumer.createJavaConsumerConnector(this.consumerConfig());
this.topic = topic;
}
private ConsumerConfig consumerConfig() {
Properties props = new Properties();
//消费者使用zk的地址获取连接
props.put("zookeeper.connect", KafkaProperties.ZK);
props.put("group.id", KafkaProperties.GROUP_ID);
props.put("zookeeper.session.timeout.ms", KafkaProperties.SESSION_TIMEOUT);
props.put("zookeeper.sync.time.ms", KafkaProperties.SYNC_TIMEOUT);
props.put("auto.commit.interval.ms", KafkaProperties.INTERVAL);
return new ConsumerConfig(props);
}
@Override
public void run() {
Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
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("Receive->[" + new String(it.next().message()) + "]");
try {
sleep(SLEEP);
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
}
5 编写启动辅助类
public class KafkaClientApp {
public static void main(String[] args) {
KafkaProducer pro = new KafkaProducer(KafkaProperties.TOPIC);
pro.start();
KafkaConsumer con = new KafkaConsumer(KafkaProperties.TOPIC);
con.start();
}
}
之后启动测试一下就好了.