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

}

}