安裝
需要在機(jī)器上配置jdk環(huán)境
下載kafka對應(yīng)版本wget wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.2/kafka_2.12-0.11.0.2.tgz
解壓后可以看到目錄
bin：包含Kafka運(yùn)行的所有腳本蔫骂，如：start/stop Zookeeper西疤，start/stop Kafka
libs：Kafka運(yùn)行的依賴庫
config：zookeeper兢交，Logger，Kafka等相關(guān)配置文件
sit-docs：Kafka相關(guān)文檔
kafka的配置方式
單節(jié)點(diǎn)-單Broker集群：只在一個(gè)節(jié)點(diǎn)上部署一個(gè)Broker
單節(jié)點(diǎn)-多Broker集群：在一個(gè)節(jié)點(diǎn)上部署多個(gè)Broker，只不過各個(gè)Broker以不同的端口啟動(dòng)
多節(jié)點(diǎn)-多Broker集群：以上兩種的組合，每個(gè)節(jié)點(diǎn)上部署一到多個(gè)Broker，且各個(gè)節(jié)點(diǎn)連接起來
這里選擇使用kafka自帶zookeeper來存儲(chǔ)集群元數(shù)據(jù)和Consumer信息。
也可以獨(dú)立部署來進(jìn)行存儲(chǔ)匆赃。
啟動(dòng)
第一步啟動(dòng)zookeeper服務(wù)
。
今缚。
啟動(dòng)成功2181端口就是zookeeper端口
可以通過修改config/zookeeper.properties 文件進(jìn)行修改
第二部啟動(dòng)kafka服務(wù)
使用kafka
通過命令新建topic
在當(dāng)前節(jié)點(diǎn)上新建一個(gè)名稱為topic1的topic
校驗(yàn)topic是否創(chuàng)建成功
topic已經(jīng)創(chuàng)建成功算柳，可以使用了。
Producer發(fā)送消息hello word!
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic1
hello world!
Consummer接受消息
image.png
下面開始編碼實(shí)現(xiàn)功能姓言。
客戶端使用lib包在服務(wù)器安裝libs目錄下
image.png
本機(jī)外調(diào)用需要修改server.properties文件
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.1.33:9092
zookeeper.connect=192.168.1.33:2181
代碼分：配置埠居、生產(chǎn)者查牌、消費(fèi)者、調(diào)用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() {}
}
生產(chǎn)者
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();
}
}
}
消費(fèi)者
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;
}
}
調(diào)用方法
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();
}
}
