1.准备3台虚拟机
安装jdk、安装zookeeper(上一篇有讲)
创建文件存放目录
mkdir -p /export/servers/
mkdir -p /export/software/
mkdir -p /export/data/
mkdir -p /export/logs/
2.下载安装包及解压
tar -zxvf kafka_2.11-1.0.0.tgz -C /export/servers/
cd /export/servers/
mv kafka_2.11-1.0.0 kafka
3.查看目录及修改配置文件
1.查看目录
1544787057518.png
2.修改配置文件
进入配置目录，查看server.properties文件
cat server.properties |grep -v "#"
通过以上命令，查看到了默认的配置文件，对默认的文件进行修改。
修改三个地方
vi server.properties
1） Borker.id
2） 数据存放的目录，注意目录如果不存在，需要新建下。
3） zookeeper的地址信息
#broker.id 标识了kafka集群中一个唯一broker。
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
#存放生产者生产的数据 数据一般以topic的方式存放
log.dirs=/export/data/kafka
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
#zk的信息
zookeeper.connect=node01:2181,node02:2181,node03:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
#可选配置
delete.topic.enable=true
host.name=192.168.72.141
4.分发配置文件及修改brokerid
将修改好的配置文件，分发到node02，node03上。
创建一个数据存放目录 /export/data/kafka
分发安装包
scp -r /export/servers/kafka/ node02:/export/servers/
scp -r /export/servers/kafka/ node03:/export/servers/
修改node02上的broker.id为1
vi /export/servers/kafka/config/server.properties
修改node03上的broker.id为2
vi /export/servers/kafka/config/server.properties
5.启动集群
cd /export/servers/kafka/bin
nohup ./kafka-server-start.sh /export/servers/kafka/config/server.properties > /dev/null 2>&1 &
输出到文件：
nohup ./kafka-server-start.sh /export/servers/kafka/config/server.properties > /export/logs/kafka/kafka.log 2>&1 &
6.自动化启动脚本
startKafka.sh
#!/bin/sh
cat /export/servers/kafka/bin/slave | while read line
do
{
echo $line
ssh $line "source /etc/profile;nohup /export/servers/kafka/bin/kafka-server-start.sh /export/servers/kafka/config/server.properties >/dev/null 2>&1 &"
}&
wait
done
stopKafka.sh
#!/bin/sh
cat /export/servers/kafka/bin/slave | while read line
do
{
echo $line
ssh $line "source /etc/profile;jps |grep Kafka |cut -c 1-5 |xargs kill -s 9 "
}&
wait
done
slave 存放的位置:/export/servers/kafka/bin/slave
node01
node02
node03
7.操作集群
创建一个订单的topic
./kafka-topics.sh --create --zookeeper node01:2181,node02:2181,node03:2181 --topic order --partitions 3 --replication-factor 2
分片数据: 三个分区 --partitions 3
每个分区的副本数据:--replication-factor 2
启动一个生产者，生产数据
./kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic order
启动一个消费者，消费数据
./kafka-console-consumer.sh --zookeeper node01:2181 --from-beginning --topic order
最新版本的命令：
./kafka-console-consumer.sh --bootstrap-server node01:9092,node02:9092,node03:9092 --topic order --from-beginning
查看topic的详细信息
./kafka-topics.sh --describe --topic order --zookeeper node01:2181,node02:2181,node03:2181
8.使用Java api运行
依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.1</version>
</dependency>
生产者的代码
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class OrderProducer {
public static void main(String[] args) {
//1. 加载生产者的配置信息
Properties props = new Properties();
//kafka集群的地址
props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
//消息的确认机制
props.put("acks", "all");
//消息的重试
props.put("retries", 0);
//消息的批量大小
props.put("batch.size", 16384);
props.put("linger.ms", 1);
//消息的批量缓存大小
props.put("buffer.memory", 33554432);
//消息的key的序列化方式
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//消息的value的序列化方式
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//2. 创建生产者
Producer<String, String> producer = new KafkaProducer<String, String>(props);
//3 生产数据
for (int i = 0; i < 100; i++) {
//一条消息
/**
* 第一个参数: topic的名称
* 第二个参数: key
* 第三个参数: value
*/
ProducerRecord<String, String> record = new ProducerRecord<String, String>("order", Integer.toString(i), Integer.toString(i));
producer.send(record);
}
//4关闭
producer.close();
}
}
消费者的代码
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class OrderConsumer {
public static void main(String[] args) {
//1. 加载配置信息
Properties props = new Properties();
//kafka集群的地址
props.put("bootstrap.servers", "node01:9092,node02:9092,node03:9092");
//消费组id(必须唯一)
props.put("group.id", "ooxx");
//消费完数据,提交消费数据的偏移量
props.put("enable.auto.commit", "true");
//提交偏移量的间隔周期
props.put("auto.commit.interval.ms", "1000");
//key和value的序列化方式
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//2. 创建消费者
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
//3. 订阅topic的消息(可以同时订阅多个topic的消息)
kafkaConsumer.subscribe(Arrays.asList("order","news"));
//4. 消费数据
while (true) {
//拉取消息 pull
ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
//遍历消息
for (ConsumerRecord<String, String> record : records) {
//获取消费的偏移量 消息的key 消息的value
int partition = record.partition();
System.out.println("partition = " + partition + " , offset = " + record.offset() + ", key = " + record.key() + ", value = " + record.value());
}
}
}
}
如果本机没有修改hosts ，用命令的方式可以发送和接收数据，用java代码发送接收不了
window系统的电脑的hosts文件中必须配置:ip和主机名的对应关系.
192.168.72.141 node01
192.168.72.142 node02
192.168.72.143 node03
