Centos 下Kafka集群的搭建

第一步

先去官网下载 kafka_2.10-0.9.0.0.tgz 并解压再进入到安装目录，也可以自己配置路径，跟名为kafka。

tar -xzfkafka_2.10-0.9.0.0.tgz
mv kafka_2.10-0.9.0.0 kafka

第二步
1.zookeeper集群搭建在hadoop0，hadoop1，hadoop2上。

2、进入到打开/ect下的hosts文件末尾
修改真加内容其中（ip和机器名根据个人实际情况修改）

vi /etc/hosts
192.168.163.128 hadoop0
192.168.163.129 hadoop1
192.168.163.130 hadoop2

第三步
启动zookeeper服务（每台机子的zookeeper都要启）

[root@hadoop0bin]# zkServer.sh start[root@hadoop1 bin]# zkServer.sh start
[root@hadoop2 bin]# zkServer.sh start

第四步

配置kafka

1、在kafka安装目录下的config目录下打开server.properties文件
修改

#其他两台机子的server.properties文件中的broker.id也要改，三台机子的broker.id不能有重复
broker.id=0#如果在同一台机子上，设置不同的端口值port=9092 host.name=hadoop0
log.dirs=/usr/softinstall/kafka/kafka-logs-0
#zookeeper集群中主机名和其端口号zookeeper.connect=hadoop0:2181,hadoop1:2181,hadoop2:2181

# Licensed to the Apache Software Foundation (ASF) under one or more# contributor license agreements.  See the NOTICE file distributed with# this work for additional information regarding copyright ownership.# The ASF licenses this file to You under the Apache License, Version 2.0# (the "License"); you may not use this file except in compliance with# the License.  You may obtain a copy of the License at# #    http://www.apache.org/licenses/LICENSE-2.0# # Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.# see kafka.server.KafkaConfig for additional details and defaults
############################# Server Basics #############################
# The id of the broker. This must be set to a unique integer for each broker.#其他两台机子的server.properties文件中的broker.id也要改，三台机子的broker.id不能有重复
broker.id=0
############################# Socket Server Settings #############################
listeners=PLAINTEXT://:9092
# The port the socket server listens on#port=9092#如果在同一台机子上，设置不同的端口值port=9092
# Hostname the broker will bind to. If not set, the server will bind to all interfaces#host.name=hadoop0host.name=hadoop0
# Hostname the broker will advertise to producers and consumers. If not set, it uses the# value for "host.name" if configured.  Otherwise, it will use the value returned from# java.net.InetAddress.getCanonicalHostName().#advertised.host.name=<hostname routable by clients>
# The port to publish to ZooKeeper for clients to use. If this is not set,# it will publish the same port that the broker binds to.#advertised.port=<port accessible by clients>
# The number of threads handling network requestsnum.network.threads=3
# The number of threads doing disk I/Onum.io.threads=8
# The send buffer (SO_SNDBUF) used by the socket serversocket.send.buffer.bytes=102400
# The receive buffer (SO_RCVBUF) used by the socket serversocket.receive.buffer.bytes=102400
# The maximum size of a request that the socket server will accept (protection against OOM)socket.request.max.bytes=104857600

############################# Log Basics #############################
# A comma seperated list of directories under which to store log fileslog.dirs=/usr/softinstall/kafka/kafka-logs-0
# The default number of log partitions per topic. More partitions allow greater# parallelism for consumption, but this will also result in more files across# the brokers.num.partitions=1
# The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.# This value is recommended to be increased for installations with data dirs located in RAID array.num.recovery.threads.per.data.dir=1
############################# Log Flush Policy #############################
# Messages are immediately written to the filesystem but by default we only fsync() to sync# the OS cache lazily. The following configurations control the flush of data to disk. # There are a few important trade-offs here:#    1. Durability: Unflushed data may be lost if you are not using replication.#    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.#    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks. # The settings below allow one to configure the flush policy to flush data after a period of time or# every N messages (or both). This can be done globally and overridden on a per-topic basis.
# The number of messages to accept before forcing a flush of data to disk#log.flush.interval.messages=10000
# The maximum amount of time a message can sit in a log before we force a flush#log.flush.interval.ms=1000
############################# Log Retention Policy #############################
# The following configurations control the disposal of log segments. The policy can# be set to delete segments after a period of time, or after a given size has accumulated.# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens# from the end of the log.
# The minimum age of a log file to be eligible for deletionlog.retention.hours=168
# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining# segments don't drop below log.retention.bytes.#log.retention.bytes=1073741824
# The maximum size of a log segment file. When this size is reached a new log segment will be created.log.segment.bytes=1073741824
# The interval at which log segments are checked to see if they can be deleted according # to the retention policieslog.retention.check.interval.ms=300000
# By default the log cleaner is disabled and the log retention policy will default to just delete segments after their retention expires.# If log.cleaner.enable=true is set the cleaner will be enabled and individual logs can then be marked for log compaction.log.cleaner.enable=false
############################# Zookeeper #############################
# 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=hadoop0:2181,hadoop1:2181,hadoop2:2181
# Timeout in ms for connecting to zookeeperzookeeper.connection.timeout.ms=6000

2、修改producer.properties文件
修改

metadata.broker.list=hadoop0:9092,hadoop1:9092,hadoop2:9092
producer.type=sync

3、修改consumer.properties文件
修改

zookeeper.connect=hadoop0:2181,hadoop1:2181,hadoop2:2181

4、拷贝kafka到hadoop1，hadoop2上。

scp -r kafka hadoop1:/usr/softinstall/scp -r kafka hadoop2:/usr/softinstall/

5、修改hadoop1，hadoop2在kafka安装目录下的config目录下打开server.properties文件

hadoop1
broker.id=1#如果在同一台机子上，设置不同的端口值port=9092 host.name=hadoop1
log.dirs=/usr/softinstall/kafka/kafka-logs-1

hadoop2
broker.id=2#如果在同一台机子上，设置不同的端口值port=9092 host.name=hadoop2
log.dirs=/usr/softinstall/kafka/kafka-logs-2

6、在hadoop0，hadoop1，hadoop2机子启动kafka服务

bin/kafka-server-start.sh config/server.properties &

[root@hadoop0 kafka]# jps3555 QuorumPeerMain6410 Kafka6462 Jps

第四步：建立一个主题

bin/kafka-topics.sh --create --zookeeper hadoop0:2181,hadoop1:2181,hadoop2:2181 --replication-factor 3 --partitions 6 --topic mytest
factor大小不能超过broker数

通过以下命令查看主题topic

[root@hadoop0 kafka]# bin/kafka-topics.sh --list --zookeeper hadoop0:2181,hadoop1:2181,hadoop2:2181my-replicated-topicmy-topicmytopic1testtest1 - marked for deletiontest2 - marked for deletiontopic1

通过下述命令可以看到该主题详情

bin/kafka-topics.sh --describe --zookeeper hadoop0:2181,hadoop1:2181,hadoop2:2181 --topic mytestTopic:mytest PartitionCount:6 ReplicationFactor:3 Configs: Topic: mytest Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2 Topic: mytest Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0 Topic: mytest Partition: 2 Leader: 2 Replicas: 2,0,1 Isr: 2,0,1 Topic: mytest Partition: 3 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1 Topic: mytest Partition: 4 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2 Topic: mytest Partition: 5 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

下面解释一下这些输出。第一行是对所有分区的一个描述，然后每个分区都会对应一行，因为我们只有一个分区所以下面就只加了一行。

leader：负责处理消息的读和写，leader是从所有节点中随机选择的.
replicas：列出了所有的副本节点，不管节点是否在服务中.
isr：是正在服务中的节点.

第五步：生产者生成消息和消费者消费消息
在hadoop0上建立生产者角色，并发送消息（其实可以是三台机子中的任何一台）

bin/kafka-console-producer.sh --broker-list hadoop1:9092 --topic mytest this is a messagethis is the second message

在hadoop2上建立消费者角色（在该终端窗口内可以看到生产者发布这消息）

[root@hadoop2 kafka]# bin/kafka-console-consumer.sh --zookeeper hadoop1:2181 --topic mytest --from-beginningthis is a messagethis is the second message

一个kafka集群就搭好了，可以作为kafka服务器了.

kafka的Java客户端环境搭建

第一步建立maven项目，添加kafka依赖包。

<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.10</artifactId> <version>0.8.2.1</version> </dependency>
 <dependency> <groupId>commons-io</groupId> <artifactId>commons-io</artifactId> <version>2.4</version> </dependency>

第二步建立producer类

package com.east.kafka;
import java.io.File;import java.io.IOException;import java.util.Collection;import java.util.Date;import java.util.List;import java.util.Properties;import org.apache.commons.io.FileUtils;
import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.ProducerConfig;
public class ProducerTest {    public static void main(String[] args) throws IOException, InterruptedException {        Properties props = new Properties();        props.put("zookeeper.connect", "hadoop0:2181,hadoop1:2181,hadoop2:2181");        props.put("serializer.class", "kafka.serializer.StringEncoder");        props.put("metadata.broker.list", "hadoop0:9092");        ProducerConfig config = new ProducerConfig(props);        Producer<String, String> producer = new Producer<String, String>(config);        while (true) {            //读取d:\\test 下的TXT文件            Collection<File> listFiles = FileUtils.listFiles(new File("d:\\test"), new String[] { "txt" }, true);            for (File file : listFiles) {                List<String> lines = FileUtils.readLines(file);                //读取文件每一行的内容                for (String line : lines) {                    Date date = new Date();                    System.out.println(date + "  send:" + line);                    KeyedMessage<String, String> message = new KeyedMessage<String, String>("my-replicated-topic",                            line);                    //生产者发送消息                    producer.send(message);                }                //更改文件名                FileUtils.moveFile(file, new File(file.getAbsolutePath() + System.currentTimeMillis()));                Thread.sleep(6000);              }        }
    }}

    }}

第三步建立Consumer类

package com.east.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.ConsumerIterator;import kafka.consumer.KafkaStream;import kafka.javaapi.consumer.ConsumerConnector;import kafka.message.MessageAndMetadata;
public class ConsumerTest extends Thread {    private final ConsumerConnector consumer;    private final String topic;    public static void main(String[] args) {        ConsumerTest consumerThread = new ConsumerTest("my-replicated-topic");        consumerThread.start();    }
    public ConsumerTest(String topic) {        System.out.println(topic);        consumer = Consumer.createJavaConsumerConnector(createConsumerConfig());        this.topic = topic;    }
    private static ConsumerConfig createConsumerConfig() {        Properties props = new Properties();        props.put("zookeeper.connect", "hadoop0:2181,hadoop1:2181,hadoop2:2181");        props.put("group.id", "0");        props.put("zookeeper.session.timeout.ms", "400000");        props.put("zookeeper.sync.time.ms", "200");        props.put("auto.commit.interval.ms", "1000");        return new ConsumerConfig(props);
    }
    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()) {            MessageAndMetadata<byte[], byte[]> next = it.next();            String message = next.message().toString();            System.out.println("recevice：" + message);        }    }}

kafka Java demo

问题kafka (kafka delete topic,marked for deletion)http://blog.csdn.net/wind520/article/details/48710043

kafka 删除topic 提示marked for deletion

[html] view plain copy

[root@logSer config]# kafka-topics.sh --delete --zookeeper localhost:2181 --topic test-group
Topic test-group is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.
[root@logSer config]# kafka-topics.sh --list --zookeeper localhost:2181
test-group - marked for deletion
test-topic
test-user-001

并没有真正删除，如果要真正删除

配置delete.topic.enable=true

配置文件在kafka\config目录

[html] view plain copy

[root@logSer config]# vi server.properties

参考连接http://my.oschina.net/u/1757031/blog/408511http://my.oschina.net/u/1757031/blog/408511http://www.bkjia.com/yjs/967259.html#top（Centos 下Kafka集群的安装，centoskafka）http://www.bkjia.com/yjs/1039755.html（kafka 安装步骤，kafka安装步骤）http://www.bkjia.com/yjs/1032939.html（kafka集群搭建，kafka集群）http://blog.csdn.net/lizhitao/article/details/39499283( apache kafka技术分享系列(目录索引))http://my.oschina.net/guol/blog/413664http://my.oschina.net/u/1024514/blog?disp=1&catalog=3282464&sort=time&p=2(重点)http://blog.csdn.net/honglei915/article/category/2383433(重点)http://my.oschina.net/u/1024514/blog/393504（kafka中级）http://my.oschina.net/crxy/blog/386513（kafka设计原理）http://blog.csdn.net/honglei915/article/details/37564329http://my.oschina.net/ydsakyclguozi/blog/381898

秒客网

Centos 下Kafka集群的搭建

Centos 下Kafka集群的搭建

相关文章