安装
需要在机器上配置jdk环境
下载kafka对应版本wget wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.2/kafka_2.12-0.11.0.2.tgz
解压后可以看到目录
image.png
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();
}
}
网友评论