网站建设 国家标准,品牌标志,网站如何做容易收录,做饮食找工作哪个网站好1、why kafka
优点 缺点kafka 吞吐量高#xff0c;对批处理和异步处理做了大量的设计#xff0c;因此Kafka可以得到非常高的性能。延迟也会高#xff0c;不适合电商场景。RabbitMQ 如果有大量消息堆积在队列中#xff0c;性能会急剧下降每秒处理几万到几十万的消息。如果…1、why kafka
优点 缺点kafka 吞吐量高对批处理和异步处理做了大量的设计因此Kafka可以得到非常高的性能。延迟也会高不适合电商场景。RabbitMQ 如果有大量消息堆积在队列中性能会急剧下降每秒处理几万到几十万的消息。如果应用要求高的性能不要选择RabbitMQ。性能RocketMQ低RocketMQ 性能比RabbitMQ高一个数量级适合电商场景。RocketMQ主要用于有序事务流计算消息推送日志流处理binlog分发等场景。每秒处理几十万的消息同时响应在毫秒级。如果应用很关注响应时间可以使用RocketMQ。
2、Broker
缓存代理(可以把Broker理解为Kafka的服务器)Kafka 集群中的一台或多台服务器统称为 broker。kafka中支持消息持久化的生产者生产消息后kafka不会直接把消息传递给消费者而是先要在broker中进行存储持久化是保存在kafka的日志文件中。
3、分区
一个消费者可以对应多个分区一个分区只能对应一个消费者。
topic分区有leader和follower。 Log的分区被分布到集群中的多个服务器上。每个服务器处理它分到的分区。 根据配置每个分区还可以复制到其它服务器作为备份容错。 每个分区有一个leader零或多个follower。Leader处理此分区的所有的读写请求而follower被动的复制数据。如果leader宕机其它的一个follower会被推举为新的leader。 一台服务器可能同时是一个分区的leader另一个分区的follower。 这样可以平衡负载避免所有的请求都只让一台或者某几台服务器处理。
4、消费者组
topic实现JMS模型中消费者组中只有一个消费者这种情况下topic的消费的offset是无序的。当单个消费者无法跟上数据生成的速度就可以增加更多的消费者分担负载每个消费者只处理部分partition的消息从而实现单个应用程序的横向伸缩。但是不要让消费者的数量多于partition的数量此时多余的消费者会空闲。此外Kafka还允许多个应用程序从同一个Topic读取所有的消息此时只要保证每个应用程序有自己的消费者组即可。
kafka为什么读写快
利用零拷贝和页面缓存技术零拷贝技术读取文件数据并发送到网络的步骤如下
将磁盘文件的数据复制到页面缓存。将数据从页面缓存直接发送到网卡从而发到网络中。
rebalance
主要是对partition的个数和group当中的consumer个数重新统计再重新对应consumer和partition的关系。一个消费者可以对应多个分区。一个分区只能对应一个消费者。
kafka producer API
生产者的分区由key决定
我们创建消息的时候必须要提供主题和消息的内容而消息的key是可选的当不指定key时默认为null。消息的key有两个重要的作用1提供描述消息的额外信息2用来决定消息写入到哪个分区所有具有相同key的消息会分配到同一个分区中。
如果key为null那么生产者会使用默认的分配器该分配器使用轮询round-robin算法来将消息均衡到所有分区。
如果key不为null而且使用的是默认的分配器那么生产者会对key进行哈希并根据结果将消息分配到特定的分区。
案例
Here is a simple example of using the producer to send records with strings containing sequential numbers as the key/value pairs.
Properties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(acks, all);props.put(retries, 0);props.put(batch.size, 16384); //默认是16kB, 每个Batch要存放batch.size大小的数据后才可以发送出去。props.put(linger.ms, 1); //一个Batch被创建之后最多过多久不管这个Batch有没有写满都必须发送出去了。props.put(buffer.memory, 33554432); //默认是32MB,KafkaProducer发送出去的消息都是先进入到客户端本地的内存缓冲里然后把很多消息收集成一个一个的Batch再发送到Broker上去的。props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer);ProducerString, String producer new KafkaProducer(props);for(int i 0; i 100; i)producer.send(new ProducerRecordString, String(my-topic, Integer.toString(i), Integer.toString(i)));producer.close();kafka consumer API
案例一手动同步提交
Properties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(group.id, test);props.put(enable.auto.commit, false);props.put(key.deserializer, org.apache.kafka.common.serialization.StringDeserializer);props.put(value.deserializer, org.apache.kafka.common.serialization.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumer(props);consumer.subscribe(Arrays.asList(foo, bar));final int minBatchSize 200;ListConsumerRecordString, String buffer new ArrayList();while (true) {ConsumerRecordsString, String records consumer.poll(100);for (ConsumerRecordString, String record : records) {buffer.add(record);}if (buffer.size() minBatchSize) {insertIntoDb(buffer);consumer.commitSync();buffer.clear();}}案例二每个partition手动同步提交
try {while(running) {ConsumerRecordsString, String records consumer.poll(Long.MAX_VALUE);for (TopicPartition partition : records.partitions()) {//拿到这个partition下面的所有数据ListConsumerRecordString, String partitionRecords records.records(partition);for (ConsumerRecordString, String record : partitionRecords) {System.out.println(record.offset() : record.value());}//通过这个partition的list获取最后一个数据的offsetlong lastOffset partitionRecords.get(partitionRecords.size() - 1).offset();consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset 1)));}}} finally {consumer.close();}Kafka文件存储
知道通过分片和索引机制找到offset的就行了。index和log文件以当前的第一条消息的offset命名。
