kafka是一个分布式消息发布和订阅系统，与zookeeper进行协调，多分区、多副本；
特性是高吞吐、可持久化、可水平扩展、支持流数据处理，具备三大功能：
- 消息系统：业务系统经常用到，常见解耦，还有削峰、异步通信、缓冲等功能。
- 存储系统
- 流式处理平台
-
Producer：生产者，发送消息的一方。
-
Consumer：消费者，接收消息的一方。
-
Broker：代理，kafka节点，一个节点就是一个kafka server进程。
-
Topic：主题，消息以主题来进行归类。
-
Partition：分区。主题的所有消息分布在不同的区中，每个分区的消息一定是不同的，分区可以分布在不同的broker中。
-
Replica：副本机制，每个分区引入多副本，leader副本和follower副本，leader副本处理读写，follower副本负责同步leader副本的数据，出现故障时，follower副本中重新选举出新的leader副本，进行故障转移。
-
Consumer Group：每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name,若不指定group name，则属于默认的group）
理解模型
kafka安装需要有jdk环境
官网下载： https://kafka.apache.org/downloads
解压tar -zxvf kafka_2.13-3.2.0.tgz配置server.properties
cd /usr/local/kafka/kafka_2.13-3.2.0/config vi server.properties
//放开listeners 主机IP listeners=PLAINTEXT://127.0.0.1:9092 //因为kafka是基于zookeeper的（新版本自带zk）,要加上zk的地址 zookeeper.connect=本服务器的ip地址:2181启动zk
./zookeeper-server-start.sh -daemon ../config/zookeeper.properties启动kafka
./kafka-server-start.sh -daemon ../config/server.propertiesJava-kafka发布订阅demo 添加maven依赖
org.apache.kafka
kafka-clients
0.10.2.0
生产者
import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Future;
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.StringSerializer;
public class Producer {
public static String topic = "test";//定义主题
public static void main(String[] args) throws InterruptedException {
Properties p = new Properties();
//kafka地址，多个地址用逗号分割
p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.181:9092");
//序列化
p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
KafkaProducer kafkaProducer = new KafkaProducer<>(p);
try {
while (true) {
String msg = "Hello," + new Random().nextInt(100);
ProducerRecord record = new ProducerRecord(topic, msg);
//kafkaProducer,send(record)可以通过返回的Future来判断是否已经发送到kafka,增强消息的可靠性
//同时也可以使用send的第二个参数来回调，通过回调判断是否发送成功。
Future send = kafkaProducer.send(record);
System.out.println("消息发送成功:" + msg);
Thread.sleep(500);
}
} finally {
kafkaProducer.close();
}
}
}
消费者
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 org.apache.kafka.common.serialization.StringDeserializer;
public class Consumer {
public static void main(String[] args) {
Properties p = new Properties();
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.181:9092");
//反序列化
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//消费组，kafka根据分组名称判断是不是同一组消费者，同一组消费者去消费一个主题的数据时，数据将在这一组消费者上面轮询
p.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
KafkaConsumer kafkaConsumer = new KafkaConsumer(p);
// 订阅消息-可订阅多个主题
kafkaConsumer.subscribe(Collections.singletonList(Producer.topic));
while (true) {
ConsumerRecords records = kafkaConsumer.poll(100);
for (ConsumerRecord record : records) {
System.out.println(String.format("topic:%s,offset:%d,消息:%s",
record.topic(), record.offset(), record.value()));
}
}
}
}
出现问题
1.启动kafka报错：ERROR Error while creating ephemeral at /brokers/ids/1, node already exists
解决：原因关闭kafka异常关闭，启动时应先启动zookeeper，再启动kafka
2.生产者发消息报错： Can’t resolve address: VM_0_15_centos:9092
解决：
1.在 server.properties，添加配置：host.name = 192.168.111.130 # 本机ip地址 2.在kafka安装服务器，hosts中添加：192.168.0.191 VM_0_15_centos暑期编程PK赛 得CSDN机械键盘等精美礼品！
