kafka介绍:
Kafka最初由LinkedIn公司开发，使用Scala语言编写，之后成为Apache项目的一部分。Kafka是一个分布式，可划分的，多订阅者，冗余备份的持久性的日志服务。它主要用于处理活跃的流式数据。在大系统中，我们经常会碰到这样的一个问题，大系统下的各个子系统需要数据高性能、低延迟的不停流转。kafka很适合处理这样的问题！
消息队列的分类：
点对点：消息生产者生产消息发送到queue中，然后消息消费者从queue中取出并且消费消息，消息被消费以后，queue中不再有储存，所以消息消费者不可能消费到已经被消费的消息。，即对消息而言，只会有一个消费者。
发布/订阅：消息生产者将消息发不到topic中，同时可以有多个消息消费者消费该消息。和点对点方式不同，发不到topic的消息会被所有订阅者消费，kafka 就是典型发布。
kafka的特点：
1、同时为发布和订阅提供高吞吐量，kafka每秒可以产生约25万的消息(50MB)，每秒能够处理55万消息(110MB)。
2、可进行持久化操作，将消息持久化到磁盘。
3、分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个。均为分布式的。
4、 消息被处理的状态是在consumer端维护，而不是在server端维护。当失败是能自动平衡。
5、支持online和offline的场景。
下面介绍一下Kafka的架构和组成：
Producer：是能够发布消息到topic的任何对象。
Consumer：消息和数据的消费者，订阅topics 并处理其发布的消息。
Consumer Group:可以并行消费Topic中的partition的消息。
Broker：缓存代理，Kafka集群中的一个kafka节点就是一个broker。
Topic: 特指Kafka 处理的消息源（feeds of messages）的不同分类。
Partition：topic物理上的分组，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列。partition中的每条消息都会被分配一个有序的 id（offset）。
Message：消息，是通信的基本单位，每个 producer 可以向一个topic（主题）发布一些消息。
kafka的安装和使用
kafka下载和相关文档地址：
修改zookeper配置文件：
dataDir=D:\tmp\kafka\zookeeper
clientPort=2181
maxClientCnxns=0
修改server配置文件：
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=D:\tmp\kafka\kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
启动服务：
先启动zookeper服务:
./bin/zookeeper-server-start.sh ./config/zookeeper.properties
再启动server服务:
./bin/kafka-server-start.sh ./config/server.properties
pom依赖：
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.10.0.1</version>
</dependency>
Config.java文件代码：
package com.lqq.demo1;
public class Config {
public final static String TOPIC = "TEST-TOPIC";
public final static String bootstrap_servers = "localhost:9092";
public final static String group_id = "jd-group";
public final static String key_serializer="org.apache.kafka.common.serialization.StringSerializer";
public final static String value_serializer="org.apache.kafka.common.serialization.StringSerializer";
public final static String key_deserializer="org.apache.kafka.common.serialization.StringDeserializer";
public final static String value_deserializer="org.apache.kafka.common.serialization.StringDeserializer";
}
生产者端:
package com.lqq.demo1;
public class KProducer {
private final Producer<String, String> producer;
public KProducer() {
Properties props = new Properties();
props.put("bootstrap.servers", Config.bootstrap_servers);
props.put("acks", "all");
props.put("retries", 1);
props.put("batch.size", 16384);
props.put("key.serializer", Config.key_deserializer);
props.put("value.serializer", Config.value_deserializer);
producer = new KafkaProducer<>(props);
}
public void produce() {
for(int i=0;i<1000;i++){
String key = String.valueOf(i);
String data = "hello kafka message " + i;
ProducerRecord<String, String> record=new ProducerRecord<String, String>(Config.TOPIC, key, data);
System.out.println("Produce record key: "+key+" value: "+data);
producer.send(record);
}
}
public void close(){
producer.close();
}
public static void main(String[] args) {
KProducer producer=new KProducer();
producer.produce();
producer.close();
}
}
指定broker和序列化类型，然后向broker发送消息。
消费者端：
package com.lqq.demo1;
public class KConsumer {
private final Consumer<String, String> consumer;
public KConsumer() {
Properties props = new Properties();
props.put("bootstrap.servers", Config.bootstrap_servers);
props.put("group.id", Config.group_id);
props.put("enable.auto.commit", "true");
props.put("client.id", "25424tg2");
props.put("heartbeat.interval.ms","1000");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("key.deserializer", Config.key_deserializer);
props.put("value.deserializer", Config.value_deserializer);
consumer = new KafkaConsumer<String, String>(props);
}
public void consume() {
consumer.subscribe(Arrays.asList(Config.TOPIC));
consumer.seekToBeginning(new ArrayList<TopicPartition>());
while(true){
ConsumerRecords<String,String> records=consumer.poll(1000);
for(ConsumerRecord<String,String> record:records){
System.out.println("Consumer record offset="+record.offset()+" key="+record.key()+" value="+record.value());
}
}
}
public void close(){
consumer.close();
}
public static void main(String[] args) {
KConsumer kConsumer=new KConsumer();
kConsumer.consume();
kConsumer.close();
}
}
