Kafka集群模式安装（二）

我们来安装Kafka的集群模式，三台机器：

192.168.131.128

192.168.131.130

192.168.131.131

Kafka集群需要依赖zookeeper，所以需要先安装好zk。

下载Kafka安装包：

Kafka_2.11-1.1.0.tgz

解压到 /usr/local/下。

进入到Kafka的config目录下：

我们看到有zk的配置文件，这是Kafka自带的zk，如果你没有安装zk，可以使用Kafka集成的zk，配置方式和单独安装是一样的。

我们默认已经安装zk，所以修改server.properties文件，大致的配置项有这些：

broker.id=0        #每个实例不一样

listeners=PLAINTEXT://192.168.131.128:9092    #改为所在主机的ip

advertised.host.name=192.168.131.128　　　　 #改为改为所在主机的ip

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/usr/local/Kafka/log                                         #需手动创建，Kafka并不会根据配置文件自动创建

num.partitions=1

num.recovery.threads.per.data.dir=1

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

zookeeper.connect=192.168.131.128:2181,192.168.131.130:2181,192.168.131.131:2181 #修改为zookeeper所在主机ip:port

zookeeper.connection.timeout.ms=6000

delete.topic.enable=true

auto.create.topics.enable=false

需要修改的地方已经标注出来了。

然后我们需要将Kafka同步到另外两台机器上：

scp -r Kafka hadoop@hadoopslaver1:/usr/local

scp -r Kafka hadoop@hadoopslaver2:/usr/local

下面我们准备启动，首先确保zk是启动的，如果没有安装可以使用Kafka的zk：

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

然后我们启动Kafka：

bin/Kafka-server-start.sh -daemon config/server.properties &

三台机器上都要执行启动操作，如果偶没有报错就是启动成功了。

接下来我们可以做一些测试。

消费端：

import java.util.Arrays;

import java.util.Properties;

import org.apache.Kafka.clients.consumer.Consumer;

import org.apache.Kafka.clients.consumer.ConsumerConfig;

import org.apache.Kafka.clients.consumer.ConsumerRecord;

import org.apache.Kafka.clients.consumer.ConsumerRecords;

import org.apache.Kafka.clients.consumer.KafkaConsumer;

import org.apache.Kafka.common.serialization.StringDeserializer;

public class Consumer {

    public static void main(String[] args) {

        Properties props = new Properties();

        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.131.128:9092,192.168.131.130:9092,192.168.131.131:9092");

        props.put(ConsumerConfig.GROUP_ID_CONFIG ,"test") ;

        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        consumer.subscribe(Arrays.asList("page_visits"));

        while (true) {

            ConsumerRecords<String, String> records = consumer.poll(100);

            for (ConsumerRecord<String, String> record : records) {

                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

            }

        }

    }

}

生产者：

import java.util.Date;

import java.util.Properties;

import java.util.Random;

import org.apache.Kafka.clients.producer.Callback;

import org.apache.Kafka.clients.producer.KafkaProducer;

import org.apache.Kafka.clients.producer.Producer;

import org.apache.Kafka.clients.producer.ProducerRecord;

import org.apache.Kafka.clients.producer.RecordMetadata;

public class Producer {

    public static void main(String[] args) {

        long events = 1;

        Random rnd = new Random();

        Properties props = new Properties();

        props.put("bootstrap.servers", "192.168.131.128:9092,192.168.131.130:9092,192.168.131.131:9092");

        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");

        //配置partitionner选择策略，可选配置

        props.put("partitioner.class", "com.rickiyang.service.Partitioner");

        Producer<String, String> producer = new KafkaProducer<>(props);

        for (long nEvents = 0; nEvents < events; nEvents++) {

            long runtime = new Date().getTime();

            String ip = "192.168.2." + rnd.nextInt(255);

            String msg = runtime + ",www.example.com," + ip;

            ProducerRecord<String, String> data = new ProducerRecord<String, String>("page_visits", ip, msg);

            producer.send(data,

                    new Callback() {

                        public void onCompletion(RecordMetadata metadata, Exception e) {

                            if(e != null) {

                                e.printStackTrace();

                            } else {

                                System.out.println("The offset of the record we just sent is: " + metadata.offset());

                            }

                        }

                    });

        }

        producer.close();

    }

}

自定义分区策略：

import java.util.List;

import java.util.Map;

import org.apache.Kafka.clients.producer.Partitioner;

import org.apache.Kafka.common.Cluster;

import org.apache.Kafka.common.PartitionInfo;

public class Partitioner implements Partitioner {

    @Override

    public void configure(Map<String, ?> configs) {

        // TODO Auto-generated method stub

    }

    @Override

    public int partition(String topic, Object key, byte[] keyBytes,

                         Object value, byte[] valueBytes, Cluster cluster) {

        int partition = 0;

        List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);

        int numPartitions = partitions.size();

        String stringKey = (String) key;

        int offset = stringKey.lastIndexOf('.');

        if (offset > 0) {

            partition = Integer.parseInt( stringKey.substring(offset+1)) % numPartitions;

        }

        return partition;

    }

    @Override

    public void close() {

        // TODO Auto-generated method stub

    }

}

我们运行一下：

Producer：

Consumer

客户端可以接受到服务端的消息的。

巴特西

Kafka集群模式安装（二）

最新文章

热门文章