1.下載

去官網下載即可

2.解壓

3.配置環境變數

/etc/profile

4.安裝zookeeper

ZooKeeper完全分散式叢集搭建

5.我先按照官方文件在單節點上用一下，先不修改配置檔案

5.1啟動zookeeper

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/zookeeper-server-start.sh config/zookeeper.properties 

5.2啟動kafka服務

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-server-start.sh config/server.properties
 

5.3建立topic

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

5.4檢視topic

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --list --zookeeper localhost:2181
test

5.5使用生產者往topic生產訊息

[email protected]
 
:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
hello
world
i
love
u
r
u
love
me
 

5.6使用消費者消費topic裡的訊息

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
hello
world
i
love
u
r
u
love
me

--from-beginning表示從頭開始接收

6.叢集搭建及使用

6.1修改配置檔案

[email protected]:/opt/kafka_2.10-0.8.1.1/config$ vi server.properties

測試用修改兩處即可

server.properties

# The id of the broker. This must be set to a unique integer for each broker.broker伺服器的ID要唯一，在drguo3設為0，drguo4設為1，drguo5設為2，不重複就行
broker.id=0
# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.下面是zookeeper連線方式
zookeeper.connect=drguo3:2181,drguo4:2181,drguo5:2181

修改後複製kafka目錄到drguo4/5

[email protected]:/opt$ scp -r kafka_2.10-0.8.1.1/ drguo4:/opt/

改一下配置檔案裡的brokerID（上面註釋中已經說了）

[email protected]:/opt$ ssh drguo4
Welcome to Ubuntu 15.10 (GNU/Linux 4.2.0-16-generic x86_64)

 * Documentation:  https://help.ubuntu.com/

Last login: Fri Apr  1 22:29:01 2016 from 192.168.80.151
[email protected]:~$ cd /opt/kafka_2.10-0.8.1.1/config/
[email protected]:/opt/kafka_2.10-0.8.1.1/config$ vi server.properties

6.2啟動zookeeper

[email protected]:/opt/kafka_2.10-0.8.1.1$ zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.8/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[email protected]:/opt/kafka_2.10-0.8.1.1$ ssh drguo4
Welcome to Ubuntu 15.10 (GNU/Linux 4.2.0-16-generic x86_64)

 * Documentation:  https://help.ubuntu.com/

Last login: Sun Apr  3 16:33:26 2016 from 192.168.80.151
[email protected]:~$ zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.8/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[email protected]:~$ exit
登出
Connection to drguo4 closed.
[email protected]:/opt/kafka_2.10-0.8.1.1$ ssh drguo5
Welcome to Ubuntu 15.10 (GNU/Linux 4.2.0-16-generic x86_64)

 * Documentation:  https://help.ubuntu.com/

Last login: Fri Apr  1 22:27:21 2016 from 192.168.80.151
[email protected]:~$ zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.8/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
[email protected]:~$ exit
登出
Connection to drguo5 closed.

6.3啟動kafka服務，三臺都要啟動（如果想讓它在後臺執行，不佔用視窗在最後加個 &）

[email protected]/4/5:/opt/kafka_2.10-0.8.1.1$ bin/kafka-server-start.sh config/server.properties

6.4建立topic

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --create --zookeeper drguo3:2181 --replication-factor 3 -partitions 1 --topic phone
Created topic "phone".
[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --list --zookeeper drguo3:2181
phone
[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --create --zookeeper drguo3:2181 --replication-factor 3 -partitions 1 --topic pc
Created topic "pc".
[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --list --zookeeper drguo3:2181
pc
phone

6.5使用生產者往topic生產訊息

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-producer.sh --broker-list drguo3:9092 --topic phone
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
hello
world
iphone
oneplus

6.6使用消費者消費topic裡的訊息

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-console-consumer.sh --zookeeper drguo3:2181 --from-beginning --topic phone
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
hello
world
iphone
oneplus

[email protected]:/opt/kafka_2.10-0.8.1.1$ bin/kafka-topics.sh --describe --zookeeper drguo3:2181 --topic phone
Topic:phone	PartitionCount:1	ReplicationFactor:3	Configs:
	Topic: phone	Partition: 0#分割槽0	Leader: 0#leader是id為0的伺服器broker	Replicas: 0,1,2#副本在0/1/2伺服器上	Isr: 0,1,2#0/1/2處於同步

7.Java客戶端

7.1生產者

package club.drguo.kafka;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class ProducerDemo {
	public static void main(String[] args) throws Exception {
		Properties props = new Properties();
		props.put("zk.connect", "drguo3:2181,drguo4:2181,drguo5:2181");
		props.put("metadata.broker.list","drguo3:9092,drguo4:9092,drguo5:9092");
		//指定序列化類
		props.put("serializer.class", "kafka.serializer.StringEncoder");
		ProducerConfig config = new ProducerConfig(props);
		Producer<String, String> producer = new Producer<String, String>(config);

		// 傳送業務訊息
		// 讀取檔案 讀取記憶體資料庫 讀socket埠
		for (int i = 1; i <= 100; i++) {
			Thread.sleep(500);
			producer.send(new KeyedMessage<String, String>("phone",
					"iphone" + i + "plus"));
		}

	}
}

7.2消費者

package club.drguo.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.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

public class ConsumerDemo {
	private static final String topic = "phone";
	private static final Integer threads = 1;

	public static void main(String[] args) {
		
		Properties props = new Properties();
		props.put("zookeeper.connect", "drguo3:2181,drguo4:2181,drguo5:2181");
		//消費者組
		props.put("group.id", "1");
		//偏移量自動重新設定
		props.put("auto.offset.reset", "smallest");

		ConsumerConfig config = new ConsumerConfig(props);
		ConsumerConnector consumer =Consumer.createJavaConsumerConnector(config);
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put(topic, threads);
		
		Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
		
		List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
		
		for(final KafkaStream<byte[], byte[]> kafkaStream : streams){
			new Thread(new Runnable() {
				@Override
				public void run() {
					for(MessageAndMetadata<byte[], byte[]> mm : kafkaStream){
						String msg = new String(mm.message());
						System.out.println(msg);
					}
				}
			
			}).start();
		
		}
	}
}