kafka消费者客户端(0.10.0.1API)

    原创
半兽人 发表于: 2016-10-20   最后更新时间: 2018-08-10  

Kafka客户端从集群中消费消息,并透明地处理kafka集群中出现故障服务器,透明地调节适应集群中变化的数据分区。也和服务器交互,平衡均衡消费者。

public class KafkaConsumer<K,V>
extends Object
implements Consumer<K,V>

消费者TCP长连接到broker来拉取消息。故障导致的消费者关闭失败,将会泄露这些连接,消费者不是线程安全的,可以查看更多关于Multi-threaded(多线程)处理的细节。

跨版本兼容性

该客户端可以与0.10.0或更新版本的broker集群进行通信。较早的版本可能不支持某些功能。例如,0.10.0broker不支持offsetsForTimes,因为此功能是在版本0.10.1中添加的。 如果你调用broker版本不可用的API时,将报 UnsupportedVersionException 异常。

偏移量和消费者的位置

kafka为分区中的每条消息保存一个偏移量(offset),这个偏移量是该分区中一条消息的唯一标示符。也表示消费者在分区的位置。例如,一个位置是5的消费者(说明已经消费了0到4的消息),下一个接收消息的偏移量为5的消息。实际上有两个与消费者相关的“位置”概念:

消费者的位置给出了下一条记录的偏移量。它比消费者在该分区中看到的最大偏移量要大一个。 它在每次消费者在调用poll(long)中接收消息时自动增长。

“已提交”的位置是已安全保存的最后偏移量,如果进程失败或重新启动时,消费者将恢复到这个偏移量。消费者可以选择定期自动提交偏移量,也可以选择通过调用commit API来手动的控制(如:commitSync 和 commitAsync)。

这个区别是消费者来控制一条消息什么时候才被认为是已被消费的,控制权在消费者,下面我们进一步更详细地讨论。

消费者组和主题订阅

Kafka的消费者组概念,通过进程池瓜分消息并处理消息。这些进程可以在同一台机器运行,也可分布到多台机器上,以增加可扩展性和容错性,相同group.id的消费者将视为同一个消费者组

分组中的每个消费者都通过subscribe API动态的订阅一个topic列表。kafka将已订阅topic的消息发送到每个消费者组中。并通过平衡分区在消费者分组中所有成员之间来达到平均。因此每个分区恰好地分配1个消费者(一个消费者组中)。所有如果一个topic有4个分区,并且一个消费者分组有只有2个消费者。那么每个消费者将消费2个分区。

消费者组的成员是动态维护的:如果一个消费者故障。分配给它的分区将重新分配给同一个分组中其他的消费者。同样的,如果一个新的消费者加入到分组,将从现有消费者中移一个给它。这被称为重新平衡分组,并在下面更详细地讨论。当新分区添加到订阅的topic时,或者当创建与订阅的正则表达式匹配的新topic时,也将重新平衡。将通过定时刷新自动发现新的分区,并将其分配给分组的成员。

从概念上讲,你可以将消费者分组看作是由多个进程组成的单一逻辑订阅者。作为一个多订阅系统,Kafka支持对于给定topic任何数量的消费者组,而不重复。

这是在消息系统中常见的功能的略微概括。所有进程都将是单个消费者分组的一部分(类似传统消息传递系统中的队列的语义),因此消息传递就像队列一样,在组中平衡。与传统的消息系统不同的是,虽然,你可以有多个这样的组。但每个进程都有自己的消费者组(类似于传统消息系统中pub-sub的语义),因此每个进程都会订阅到该主题的所有消息。

此外,当分组重新分配自动发生时,可以通过ConsumerRebalanceListener通知消费者,这允许他们完成必要的应用程序级逻辑,例如状态清除,手动偏移提交等。有关更多详细信息,请参阅Kafka存储的偏移

它也允许消费者通过使用assign(Collection)手动分配指定分区,如果使用手动指定分配分区,那么动态分区分配和协调消费者组将失效。

发现消费者故障

订阅一组topic后,当调用poll(long)时,消费者将自动加入到组中。只要持续的调用poll,消费者将一直保持可用,并继续从分配的分区中接收消息。此外,消费者向服务器定时发送心跳。 如果消费者崩溃或无法在session.timeout.ms配置的时间内发送心跳,则消费者将被视为死亡,并且其分区将被重新分配。

还有一种可能,消费可能遇到“活锁”的情况,它持续的发送心跳,但是没有处理。为了预防消费者在这种情况下一直持有分区,我们使用max.poll.interval.ms活跃检测机制。 在此基础上,如果你调用的poll的频率大于最大间隔,则客户端将主动地离开组,以便其他消费者接管该分区。 发生这种情况时,你会看到offset提交失败(调用commitSync()引发的CommitFailedException)。这是一种安全机制,保障只有活动成员能够提交offset。所以要留在组中,你必须持续调用poll。

消费者提供两个配置设置来控制poll循环:

  1. max.poll.interval.ms:增大poll的间隔,可以为消费者提供更多的时间去处理返回的消息(调用poll(long)返回的消息,通常返回的消息都是一批)。缺点是此值越大将会延迟组重新平衡。

  2. max.poll.records:此设置限制每次调用poll返回的消息数,这样可以更容易的预测每次poll间隔要处理的最大值。通过调整此值,可以减少poll间隔,减少重新平衡分组的

对于消息处理时间不可预测地的情况,这些选项是不够的。 处理这种情况的推荐方法是将消息处理移到另一个线程中,让消费者继续调用poll。 但是必须注意确保已提交的offset不超过实际位置。另外,你必须禁用自动提交,并只有在线程完成处理后才为记录手动提交偏移量(取决于你)。 还要注意,你需要pause暂停分区,不会从poll接收到新消息,让线程处理完之前返回的消息(如果你的处理能力比拉取消息的慢,那创建新线程将导致你机器内存溢出)。

示例

这个消费者API提供了灵活性,以涵盖各种消费场景,下面是一些例子来演示如何使用它们。

自动提交偏移量

这是个【自动提交偏移量】的简单的kafka消费者API。

  Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "true");
     props.put("auto.commit.interval.ms", "1000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
             System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
     }

设置enable.auto.commit,偏移量由auto.commit.interval.ms控制自动提交的频率。

集群是通过配置bootstrap.servers指定一个或多个broker。不用指定全部的broker,它将自动发现集群中的其余的borker(最好指定多个,万一有服务器故障)。

在这个例子中,客户端订阅了主题foobar。消费者组叫test

broker通过心跳机器自动检测test组中失败的进程,消费者会自动ping集群,告诉进群它还活着。只要消费者能够做到这一点,它就被认为是活着的,并保留分配给它分区的权利,如果它停止心跳的时间超过session.timeout.ms,那么就会认为是故障的,它的分区将被分配到别的进程。

这个deserializer设置如何把byte转成object类型,例子中,通过指定string解析器,我们告诉获取到的消息的key和value只是简单个string类型。

手动控制偏移量

不需要定时的提交offset,可以自己控制offset,当消息认为已消费过了,这个时候再去提交它们的偏移量。这个很有用的,当消费的消息结合了一些处理逻辑,这个消息就不应该认为是已经消费的,直到它完成了整个处理。

Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "false");
     props.put("auto.commit.interval.ms", "1000");
     props.put("session.timeout.ms", "30000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     final int minBatchSize = 200;
     List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
         for (ConsumerRecord<String, String> record : records) {
             buffer.add(record);
         }
         if (buffer.size() >= minBatchSize) {
             insertIntoDb(buffer);
             consumer.commitSync();
             buffer.clear();
         }
     }

在这个例子中,我们将消费一批消息并将它们存储在内存中。当我们积累足够多的消息后,我们再将它们批量插入到数据库中。如果我们设置offset自动提交(之前说的例子),消费将被认为是已消费的。这样会出现问题,我们的进程可能在批处理记录之后,但在它们被插入到数据库之前失败了。

为了避免这种情况,我们将在相应的记录插入数据库之后再手动提交偏移量。这样我们可以准确控制消息是成功消费的。提出一个相反的可能性:在插入数据库之后,但是在提交之前,这个过程可能会失败(即使这可能只是几毫秒,这是一种可能性)。在这种情况下,进程将获取到已提交的偏移量,并会重复插入的最后一批数据。这种方式就是所谓的“至少一次”保证,在故障情况下,可以重复。

如果您无法执行这些操作,可能会使已提交的偏移超过消耗的位置,从而导致缺少记录。 使用手动偏移控制的优点是,您可以直接控制记录何时被视为“已消耗”。

注意:使用自动提交也可以“至少一次”。但是要求你必须下次调用poll(long)之前或关闭消费者之前,处理完所有返回的数据。如果操作失败,这将会导致已提交的offset超过消费的位置,从而导致丢失消息。使用手动控制offset的有点是,你可以直接控制消息何时提交。、

上面的例子使用commitSync表示所有收到的消息为”已提交",在某些情况下,你可以希望更精细的控制,通过指定一个明确消息的偏移量为“已提交”。在下面,我们的例子中,我们处理完每个分区中的消息后,提交偏移量。

try {
         while(running) {
             ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
             for (TopicPartition partition : records.partitions()) {
                 List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                 for (ConsumerRecord<String, String> record : partitionRecords) {
                     System.out.println(record.offset() + ": " + record.value());
                 }
                 long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                 consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
             }
         }
     } finally {
       consumer.close();
     }

注意:已提交的offset应始终是你的程序将读取的下一条消息的offset。因此,调用commitSync(offsets)时,你应该加1个到最后处理的消息的offset。

订阅指定的分区

在前面的例子中,我们订阅我们感兴趣的topic,让kafka提供给我们平分后的topic分区。但是,在有些情况下,你可能需要自己来控制分配指定分区,例如:

  • 如果这个消费者进程与该分区保存了某种本地状态(如本地磁盘的键值存储),则它应该只能获取这个分区的消息。

  • 如果消费者进程本身具有高可用性,并且如果它失败,会自动重新启动(可能使用集群管理框架如YARN,Mesos,或者AWS设施,或作为一个流处理框架的一部分)。 在这种情况下,不需要Kafka检测故障,重新分配分区,因为消费者进程将在另一台机器上重新启动。

要使用此模式,,你只需调用assign(Collection)消费指定的分区即可:

     String topic = "foo";
     TopicPartition partition0 = new TopicPartition(topic, 0);
     TopicPartition partition1 = new TopicPartition(topic, 1);
     consumer.assign(Arrays.asList(partition0, partition1));

一旦手动分配分区,你可以在循环中调用poll(跟前面的例子一样)。消费者分组仍需要提交offset,只是现在分区的设置只能通过调用assign修改,因为手动分配不会进行分组协调,因此消费者故障不会引发分区重新平衡。每一个消费者是独立工作的(即使和其他的消费者共享GroupId)。为了避免offset提交冲突,通常你需要确认每一个consumer实例的gorupId都是唯一的。

注意,手动分配分区(即,assgin)和动态分区分配的订阅topic模式(即,subcribe)不能混合使用。

offset存储在其他地方

消费者可以不使用kafka内置的offset仓库。可以选择自己来存储offset。要注意的是,将消费的offset和结果存储在同一个的系统中,用原子的方式存储结果和offset,但这不能保证原子,要想消费是完全原子的,并提供的“正好一次”的消费保证比kafka默认的“至少一次”的语义要更高。你需要使用kafka的offset提交功能。

这有结合的例子。

  • 如果消费的结果存储在关系数据库中,存储在数据库的offset,让提交结果和offset在单个事务中。这样,事物成功,则offset存储和更新。如果offset没有存储,那么偏移量也不会被更新。

  • 如果offset和消费结果存储在本地仓库。例如,可以通过订阅一个指定的分区并将offset和索引数据一起存储来构建一个搜索索引。如果这是以原子的方式做的,常见的可能是,即使崩溃引起未同步的数据丢失。索引程序从它确保没有更新丢失的地方恢复,而仅仅丢失最近更新的消息。

每个消息都有自己的offset,所以要管理自己的偏移,你只需要做到以下几点:

  • 配置 enable.auto.commit=false

  • 使用提供的 ConsumerRecord 来保存你的位置。

  • 在重启时用 seek(TopicPartition, long) 恢复消费者的位置。

当分区分配也是手动完成的(像上文搜索索引的情况),这种类型的使用是最简单的。 如果分区分配是自动完成的,需要特别小心处理分区分配变更的情况。可以通过调用subscribe(Collection,ConsumerRebalanceListener)subscribe(Pattern,ConsumerRebalanceListener)中提供的ConsumerRebalanceListener实例来完成的。例如,当分区向消费者获取时,消费者将通过实现ConsumerRebalanceListener.onPartitionsRevoked(Collection)来给这些分区提交它们offset。当分区分配给消费者时,消费者通过ConsumerRebalanceListener.onPartitionsAssigned(Collection)为新的分区正确地将消费者初始化到该位置。

ConsumerRebalanceListener的另一个常见用法是清除应用已移动到其他位置的分区的缓存。

控制消费的位置

大多数情况下,消费者只是简单的从头到尾的消费消息,周期性的提交位置(自动或手动)。kafka也支持消费者去手动的控制消费的位置,可以消费之前的消息也可以跳过最近的消息。

有几种情况,手动控制消费者的位置可能是有用的。

一种场景是对于时间敏感的消费者处理程序,对足够落后的消费者,直接跳过,从最近的消费开始消费。

另一个使用场景是本地状态存储系统(上一节说的)。在这样的系统中,消费者将要在启动时初始化它的位置(无论本地存储是否包含)。同样,如果本地状态已被破坏(假设因为磁盘丢失),则可以通过重新消费所有数据并重新创建状态(假设kafka保留了足够的历史)在新的机器上重新创建。

kafka使用seek(TopicPartition, long)指定新的消费位置。用于查找服务器保留的最早和最新的offset的特殊的方法也可用(seekToBeginning(Collection) 和 seekToEnd(Collection))。

消费者流量控制

如果消费者分配了多个分区,并同时消费所有的分区,这些分区具有相同的优先级。在一些情况下,消费者需要首先消费一些指定的分区,当指定的分区有少量或者已经没有可消费的数据时,则开始消费其他分区。

例如流处理,当处理器从2个topic获取消息并把这两个topic的消息合并,当其中一个topic长时间落后另一个,则暂停消费,以便落后的赶上来。

kafka支持动态控制消费流量,分别在future的poll(long)中使用pause(Collection)resume(Collection) 来暂停消费指定分配的分区,重新开始消费指定暂停的分区。

多线程处理

Kafka消费者不是线程安全的。所有网络I/O都发生在进行调用应用程序的线程中。用户的责任是确保多线程访问正确同步的。非同步访问将导致ConcurrentModificationException。

此规则唯一的例外是wakeup(),它可以安全地从外部线程来中断活动操作。在这种情况下,将从操作的线程阻塞并抛出一个WakeupException。这可用于从其他线程来关闭消费者。 以下代码段显示了典型模式:

public class KafkaConsumerRunner implements Runnable {
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final KafkaConsumer consumer;

     public void run() {
         try {
             consumer.subscribe(Arrays.asList("topic"));
             while (!closed.get()) {
                 ConsumerRecords records = consumer.poll(10000);
                 // Handle new records
             }
         } catch (WakeupException e) {
             // Ignore exception if closing
             if (!closed.get()) throw e;
         } finally {
             consumer.close();
         }
     }

     // Shutdown hook which can be called from a separate thread
     public void shutdown() {
         closed.set(true);
         consumer.wakeup();
     }
 }

在单独的线程中,可以通过设置关闭标志和唤醒消费者来关闭消费者。

      closed.set(true);
     consumer.wakeup();

我们没有多线程模型的例子。但留下几个操作可用来实现多线程处理消息。

  1. 每个线程一个消费者

    每个线程自己的消费者实例。这里是这种方法的优点和缺点:

    • PRO: 这是最容易实现的
    • PRO: 因为它不需要在线程之间协调,所以通常它是最快的。
    • PRO: 它按顺序处理每个分区(每个线程只处理它接受的消息)。
    • CON: 更多的消费者意味着更多的TCP连接到集群(每个线程一个)。一般kafka处理连接非常的快,所以这是一个小成本。
    • CON: 更多的消费者意味着更多的请求被发送到服务器,但稍微较少的数据批次可能导致I/O吞吐量的一些下降。
    • CON: 所有进程中的线程总数受到分区总数的限制。
  2. 解耦消费和处理

    另一个替代方式是一个或多个消费者线程,它来消费所有数据,其消费所有数据并将ConsumerRecords实例切换到由实际处理记录处理的处理器线程池来消费的阻塞队列。这个选项同样有利弊:

    • PRO: 可扩展消费者和处理进程的数量。这样单个消费者的数据可分给多个处理器线程来执行,避免对分区的任何限制。
    • CON: 跨多个处理器的顺序保证需要特别注意,因为线程是独立的执行,后来的消息可能比遭到的消息先处理,这仅仅是因为线程执行的运气。如果对排序没有问题,这就不是个问题。
    • CON: 手动提交变得更困难,因为它需要协调所有的线程以确保处理对该分区的处理完成。

这种方法有多种玩法,例如,每个处理线程可以有自己的队列,消费者线程可以使用TopicPartitionhash到这些队列中,以确保按顺序消费,并且提交也将简化。



您需要解锁本帖隐藏内容请: 点击这里
本帖隐藏的内容




上一条: kafka消费者API
下一条: Kafka Streams API

  • 大佬,我现在一个topic有三个分区,然后只启动了一个消费者用指定分区的方式去消费其中一个分区,可以正常消费,但是一直在报错:

    [main] ERROR org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1-1, groupId=1] Offset commit failed on partition testTopic-0 at offset 137: The coordinator is not aware of this member.
    [main] WARN org.apache.kafka.clients.consumer.internals.ConsumerCoordinator - [Consumer clientId=consumer-1-1, groupId=1] Asynchronous auto-commit of offsets {testTopic-0=OffsetAndMetadata{offset=137, leaderEpoch=0, metadata=''}} failed: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
    
    • 嗯,你指定分区的方式消费,但是你提交offset的时候,消费者组正在重新平衡,所以导致你无法提交成功。
      所以要check一下客户端代码,指定分区就不能自动分配消费者了,就是这个问题了。

        还有手动指定分区,如果一个消费者组中有3个消费者,一个topic有3个分区,其中消费者A手动指定了分区p1,那么p1这个分区还会被分配给剩下的两个消费者

        大佬。这里的“消费者的位置” 和 “已提交的位置”,一个值得是消费者本地已经消费的位置,而已提交的位置是提交到服务器的位置,所以这两个值可能存在偏差,即消费者的位置大于已提交的位置。比如消费者的位置已经到100了,但是已提交的位置才50,那么消费者故障之后重启,依旧是从50这个位置开始消费的吗

        大佬您好,问个问题啊,offset 手动提交 是不是 poll() 了记录就 提交更新offset了。不会管记录的处理。 如果处理失败,这不就是 ”最多一次” 了嘛

        你好,看完文章后受益匪浅。我最近遇到一个需要消费指定时间段数据的问题。比如今天凌晨1-2点的数据丢了,我需要重新消费1-2点的数据,发送出去。按照网上例子写了,也能实现效果,但会导致其他主题的消费停止。具体流程是这样,终端上传数据,数据来源包含两部分,一个是来自时间主题的指令信息,一个是来自工况主题要处理的数据信息,我消费完了数据之后发送到两个主题,一个是自己的备份主题,一个是根据指令发送特定时间段的数据到客户的主题。假如客户因为服务器原因没有收到我发送的数据,我就需要从自己的备份主题里拉取指定时间段的数据,发给他们。我拉取指定时间的数据,是用时间戳实现的。先获取topicPartition,然后用assign方法订阅。seek设置消费的分区的偏移量。然后while循环执行poll方法,判断满足1-2点的数据,将其发送出去。break条件是超过2.30了就停止消费。这么写完之后,我模拟一条时间指令,能够从备份把指定时间段内的数据拉出来,但是我消费工况主题的进程就会停止。查看卡夫卡的ui界面,发现消费者偏移量也没动,所以怀疑是出了问题。我看到你有说过,按照时间戳之后,消费者都是按照指定分区对应时间戳的Offset开始消费的。卡了两天,从网上也搜了不少的信息,还是没有头绪。目前比较怀疑的地方是,我设置了手动提交偏移量,是不是每发送一条消息都要手动提交一下。还是说我消费工况主题的方式也要改成assign才可以互不影响呢。代码有些长就没贴,不知道我是否将问题描述清楚,还请大佬指教。如有需要,我将代码贴出

        • 如果你需要重新读取1-2点的数据,这个”小程序“的消费者组名不能和现有正在运行的消费者组名相同。从你的描述来看,因为你通过相同的消费者组名重置了你的offset,就导致你每次都从对应时间戳开始消费。

          所以重新读取丢失的消息,不要跟线上任何一个消费者组名相同,你只是重读了这些消息,拿到这些消息怎么处理,是你另外的逻辑了。


          问题最好在问题专区里提问,可以描述的清楚一些
          https://www.orchome.com/kafka/issues

            • 打扰了,再请教一下。如果采用了时间戳来控制消费,那么在提交 偏移量的时候,是自动提交设置auto comit为true好,还是手动提交偏移量comitSync,如果手动提交,应该同步还是异步呢

                为消费者指定分区后,消费者生成的 consumerCoordinator 是否不再起作用了,即使为消费者指定了group.id?

                • 就是独立consumer情景,不为消费者配置group.id,就没有consumerCoordinator,那消费端就无法提供重平衡功能。
                  那为消费者assign固定分区时,消费端不会去为其做重平衡,即使做了group.id配置,在创建consumer时生成consumerCoordinator了,这个也起不到作用了。———我想表达的意思,当为消费者固定分区后,即使消费者提供了consumerCoordinator,只是白白浪费了,这种情况下无需做group.id配置,对吧。

                    你好,我可以问下poll拉取的消息数不均匀是什么问题导致的吗?
                    具体配置是这样:max.poll.records=3000 手动提交 一般情况下基本都是2000多条消息 个别出现百条或者几十条的情况 每个消息大小是7k

                    大佬你好,我想问下,在命令行的示例中是topic对应队列的,而在java api的示例中则是topic下还有key和value的内容,这些之间有什么区别吗?

                    你好,能帮我看下是什么错吗?

                    [2019-08-27 10:35:45,023] WARN [Producer clientId=console-producer] Bootstrap broker 10.11.111.38:9092 (id: -1 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
                    

                    生产者发送给kafka消息时显示kafka异步发送组件异常又是什么原因啊?

                    您好!为什么我用API写的消费者,却获取不到record,生产者可以发布record,也没出现异常

                    • 解决了,poll(100)的设置太短了。大佬,还有一问题,我设置了自动提交偏移量,为什么轮询时会一直消费相同的信息,还有auto.offset.reset=lastet是不重复消费吗?