kafka的配置和使用

安裝
須要在機器上配置jdk環境
下載kafka對應版本wget wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.2/kafka_2.12-0.11.0.2.tgz
解壓後能夠看到目錄

 

 

bin：包含Kafka運行的全部腳本，如：start/stop Zookeeper，start/stop Kafka
libs：Kafka運行的依賴庫
config：zookeeper，Logger，Kafka等相關配置文件
sit-docs：Kafka相關文檔

kafka的配置方式
單節點-單Broker集羣：只在一個節點上部署一個Broker
單節點-多Broker集羣：在一個節點上部署多個Broker，只不過各個Broker以不一樣的端口啓動
多節點-多Broker集羣：以上兩種的組合，每一個節點上部署一到多個Broker，且各個節點鏈接起來

這裏選擇使用kafka自帶zookeeper來存儲集羣元數據和Consumer信息。
也能夠獨立部署來進行存儲。
啓動
第一步啓動zookeeper服務

 

image.png

。
。

 

image.png

 

啓動成功2181端口就是zookeeper端口
能夠經過修改config/zookeeper.properties 文件進行修改
第二部啓動kafka服務

 

image.png

使用kafka
經過命令新建topic

 

image.png

 

在當前節點上新建一個名稱爲topic1的topic
校驗topic是否建立成功

 

image.png

topic已經建立成功，能夠使用了。
Producer發送消息hello word!
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1

hello world!
Consummer接受消息

 

image.png

 

下面開始編碼實現功能。
客戶端使用lib包在服務器安裝libs目錄下

 

image.png

本機外調用須要修改server.properties文件
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.1.33:9092
zookeeper.connect=192.168.1.33:2181

代碼分：配置、生產者、消費者、調用main方法4部分組成
配置文件
package com.main;

public class KafkaProperties {
public static final String TOPIC = "topic1";
public static final String KAFKA_SERVER_URL = "192.168.1.33";
public static final int KAFKA_SERVER_PORT = 9092;
public static final int KAFKA_CONSUMER_PORT=2181;
public static final int KAFKA_PRODUCER_BUFFER_SIZE = 64 * 1024;
public static final int CONNECTION_TIMEOUT = 1000;
public static final String CLIENT_ID = "SimpleConsumerDemoClient";

private KafkaProperties() {}

}
生產者
package com.main;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;

public class Producer extends Thread{
private final KafkaProducer<Integer, String> producer;
private final String topic;
private final Boolean isAsync;

public Producer(String topic, Boolean isAsync) {
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
    props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    producer = new KafkaProducer<Integer, String>(props);
    this.topic = topic;
    this.isAsync = isAsync;
}

public void run() {
    int messageNo = 1;
    while (true) {
        String messageStr = "Message_" + messageNo;
        long startTime = System.currentTimeMillis();
        if (isAsync) { // Send asynchronously
            producer.send(new ProducerRecord<Integer, String>(topic,
                messageNo,
                messageStr), new DemoCallBack(startTime, messageNo, messageStr));
            if(messageNo==100){
                break;
            }
        } else { // Send synchronously
            try {
                producer.send(new ProducerRecord<Integer, String>(topic,
                    messageNo,
                    messageStr)).get();
                System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")");
            } catch (ExecutionException e) {
                e.printStackTrace();
            }catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        ++messageNo;
    }
}

}
class DemoCallBack implements Callback {

private final long startTime;
private final int key;
private final String message;

public DemoCallBack(long startTime, int key, String message) {
    this.startTime = startTime;
    this.key = key;
    this.message = message;
}

/**
 * A callback method the user can implement to provide asynchronous handling of request completion. This method will
 * be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
 * non-null.
 *
 * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
 *                  occurred.
 * @param exception The exception thrown during processing of this record. Null if no error occurred.
 */
public void onCompletion(RecordMetadata metadata, Exception exception) {
    long elapsedTime = System.currentTimeMillis() - startTime;
    if (metadata != null) {
        System.out.println(
            "message(" + key + ", " + message + ") sent to partition(" + metadata.partition() +
                "), " +
                "offset(" + metadata.offset() + ") in " + elapsedTime + " ms");
    } else {
        exception.printStackTrace();
    }
}

}
消費者
package com.main;

import java.util.Collections;
import java.util.Properties;

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 kafka.utils.ShutdownableThread;

public class Consumer extends ShutdownableThread{
private final KafkaConsumer<Integer, String> consumer;
private final String topic;

public Consumer(String topic) {
    super("KafkaConsumerExample", false);
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    consumer = new KafkaConsumer<Integer,String>(props);
    this.topic = topic;
}

@Override
public void doWork() {
    System.out.println("doWork");
    consumer.subscribe(Collections.singletonList(this.topic));
    ConsumerRecords<Integer, String> records = consumer.poll(1000);
    for (ConsumerRecord<Integer, String> record : records) {
        System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
    }
}

@Override
public String name() {
    return null;
}

@Override
public boolean isInterruptible() {
    return false;
}

}
調用方法
package com.main;

public class KafkaConsumerProducerDemo {
public static void main(String[] args) {
boolean isAsync = true;
Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync);
producerThread.start();

Consumer consumerThread = new Consumer(KafkaProperties.TOPIC);
    consumerThread.start();

}

}