kafka---部署

Kafka是一种分布式的消息系统。本文基于0.9.0版本，利用自带的zookeeper，部署了3个broker节点的kafka集群。

下载、配置

下载

wget http://mirrors.cnnic.cn/apache/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz
tar -xzf kafka_2.11-0.9.0.0.tgz
cd kafka_2.11-0.9.0.0

启动zookeeper

config/zookeeper.properties,配置可以改数据目录，端口可以不改

dataDir=data/zookeeper              #数据目录，默认/tmp/zookeeper
clientPort=2181                     #默认2181

启动zk

nohup bin/zookeeper-server-start.sh config/zookeeper.properties >> zk.log &

启动kafka-broker

broker配置

配置可以改一下日志目录，端口冲突了可以改一下端口。

kafka-broker配置：config/server.properties

log.dirs=data/kafka-logs            #数据目录，默认/tmp/kafka-logs
broker.id=0
listeners=PLAINTEXT://:9092         #默认PLAINTEXT://:9092
port=9092                           #端口默认为9092
zookeeper.connect=localhost:2181    #zk连接信息
delete.topic.enable=true            #在最后添加：可以删除topic

另外两个broker配置

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

config/server-1.properties

broker.id=1
port=9093
listeners=PLAINTEXT://:9093
log.dir=data/kafka-logs-1

config/server-2.properties

broker.id=2
port=9094
listeners=PLAINTEXT://:9094
log.dir=data/kafka-logs-2
``` 

broker启动

nohup bin/kafka-server-start.sh config/server.properties >> broker1.log &
nohup bin/kafka-server-start.sh config/server-1.properties >> broker2.log &
nohup bin/kafka-server-start.sh config/server-2.properties >> broker3.log &

## 创建topic
- 创建topic：

bin/kafka-topics.sh –create –zookeeper localhost:2181 –partitions 3 –topic test3partitions –replication-factor 1 #包含3个分区，1个副本

为Topic创建分区时，分区数最好是broker数量的整数倍,这样才能是一个Topic的分区均匀的分布在整个Kafka集群中。  

- 查看已经创建的topic：

bin/kafka-topics.sh –list –zookeeper localhost:2181

- 查看某个topic：

bin/kafka-topics.sh –describe –zookeeper localhost:2181 –topic test3partitions

- 删除topic：

bin/kafka-topics.sh –delete –topic test3partitions –zookeeper localhost:2181

## producer
在一终端A启动producer

bin/kafka-console-producer.sh –broker-list localhost:9092,localhost:9093,localhost:9094 –topic test3partitions

## consumer
config/consumer.properties修改：

zookeeper.connect=localhost:2181 #zookeeper，后面以host:port,host port,….
group.id=test-consumer-group #消费者group-id

在另外的终端B、C、D启动consumer：

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic test3partitions

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –consumer.config config/consumer.properties –topic test3partitions

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –consumer.config config/consumer.properties –topic test3partitions

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –consumer.config config/consumer.properties –topic test3partitions

后面加 --from-beginning表示从头开始收消息

其中，B终端不配置group-id,将会为console-consumer-，加上随机数字。C、D、E终端读取配置文件中的group.id，为tet-consumer-group。

### 测试
在producer终端输入消息从1-20。可以看到B终端会输出1-20全部消息，图中B所示。而C、D、E终端由于属于同一个Consumer Group，partitions数量等于consumer，每个consumer消费了一个partition里的消息。图中为C、D、E。

将C终端断开，剩下B、D、E去消费消息。B终端还是会输出1-20全部消息，图中为B1所示。而D、E属于同一个Consumer Group，且consumer数量少于partition数，可以看到D消费了两个partition中的数据，见图中D1所示。    
![image](http://note.youdao.com/yws/public/resource/027eae2659928c104735afa6fe77369a/xmlnote/A6002CC243CC471FBF51E6BFDE0E1BA2/23687)  

## Java api 测试
利用Java api自己写一个与上面测试用例相同的例子。

Producer

public class NewProducerSimple {
public static void main(String [] args) {
Properties props = new Properties();
props.put(“bootstrap.servers”, “localhost:9092,localhost:9093,localhost:9094”);
props.put(“acks”, “all”);
props.put(“retries”, 0);
props.put(“batch.size”, 16384);
props.put(“linger.ms”, 1);
props.put(“buffer.memory”, 33554432);
props.put(“key.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
props.put(“value.serializer”, “org.apache.kafka.common.serialization.StringSerializer”);
props.put(“partitioner.class”,”com.youdao.newClient.SimplePartitioner”);
Producer producer = new KafkaProducer<>(props);
for(int i = 0; i < 20; i++)
producer.send(new ProducerRecord(“test3partitions”, Integer.toString(i), Integer.toString(i)));

    producer.close();
}

}

分区规则，key.hash%3

public class SimplePartitioner implements Partitioner {

@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    int partitionNum = 3;
    String k = (String)key;
    int partition = Math.abs(k.hashCode()) % partitionNum;
    return partition;
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> map) {
}

}

Consumer,同样，可以启动3个相同的consumer

public class NewConsumerSimple {
public static void main(String [] args) {
Properties props = new Properties();
props.put(“bootstrap.servers”, “localhost:9092,localhost:9093,localhost:9094”);
props.put(“group.id”, “test”);
props.put(“enable.auto.commit”, “true”);
props.put(“auto.commit.interval.ms”, “1000”);
props.put(“key.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
props.put(“value.deserializer”, “org.apache.kafka.common.serialization.StringDeserializer”);
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList(“test3partitions”));
while (true) {
ConsumerRecords records = consumer.poll(100);
for (ConsumerRecord record : records)
System.out.printf(“offset = %d, key = %s, value = %s%n”, record.offset(), record.key(), record.value());
}
}
}
```