搭建网站服务器需要什么配置,wap网站快速开发,slider revolution wordpress,哪个网站用div做的好文章目录 消费者回调如何抽象callBack消息#xff1f;为什么要设置serverId?如何消费callBack消息#xff1f; 定时重试消息失败表的设计重试逻辑设计 理解Rabalance通过日志来理解rebalance 参考资料结语示例源码仓库 消费者回调
有些邮件发送成功之后#xff0c;需要执行… 文章目录 消费者回调如何抽象callBack消息为什么要设置serverId?如何消费callBack消息 定时重试消息失败表的设计重试逻辑设计 理解Rabalance通过日志来理解rebalance 参考资料结语示例源码仓库 消费者回调
有些邮件发送成功之后需要执行后续逻辑例如更新数据库等。那么我们这时需要将Message Server变成生产者 向Kafak中投递callBack消息Business Server 此时是消费者 消费callBack消息。
如何抽象callBack消息
callBack的逻辑根据业务场景相关如何在保证满足不同callBack业务逻辑的同时还满足callBack消息格式的统一呢 我们使用反射来实现这一目的
JsonDeserialize(builder CallbackMetaData.CallbackMetaDataBuilder.class)
JsonInclude(JsonInclude.Include.NON_NULL)
Getter
Builder
ToString
public class CallbackMetaData implements Serializable {JsonProperty(id)private String messageId;/*** 应该是hostName, InetAddress.getLocalHost().getHostName()*/JsonProperty(serverId)private String serverId;JsonProperty(className)private String className;/*** this string is the json string of the instance of the class, generated by Jackson.* for example:* className instance new className();* objectMapper.writeValueAsString(instance);*/JsonProperty(instanceJsonStr)private String instanceJsonStr;JsonProperty(methodName)private String methodName;JsonProperty(arguments)private Object[] arguments;}上述内容会作为邮件消息的一部分发送给Message Server. 当Message Server发送完邮件之后 会检查是否包含callback消息如果包含则将CallbackMetaData发送到相关topic
为什么要设置serverId?
有两点原因
代码是滚动部署的为了兼容性必须是回调消息是由哪台Business Server携带的就该回调到哪台Business Server。一开始我们的思路是使用同一个topic所有Business Server都订阅该topic并且使用不同的消费者组以达到广播消息的目的。这样在消费消息时通过判断当前消息的serverId是否当前server如果是则消费如果不是则直接提交offset。但是后来我们发现这样设计会使得所有服务器每时每刻都在消费消息即使该消息不是当前服务器的。改进后的设计是每个服务器都有自己的callback topic 只消费自己的callback topic下的消息即可。 callback topic的名字是callback-serverId
serverId使用机器的hostName而不是IP, 因为IP有可能会变。
如何消费callBack消息
其他消费者相关的代码我不再赘述请参考上一篇博文的详细内容 执行消费逻辑的代码就3行
ConsumerRecordsString, CallbackMetaData records consumer.poll(Duration.ofSeconds(10));
records.forEach(each - {Class? destClass;try {// 核心消费代码, 通过反射调用目标方法destClass Class.forName(each.value().getClassName());Object instance objectMapper.readValue(each.value().getInstanceJsonStr(), destClass);MethodUtils.invokeMethod(instance, true, each.value().getMethodName(), each.value().getArguments());} catch (Exception e) {e.printStackTrace();}
});定时重试
前两篇博文提到无论生产者还是消费者最终重试N次之后依旧失败的我们会把消息存储到数据库以便后期通过定时任务进行重试。为了减轻业务服务器的负担所有失败消息的重试都由Message Server负责。
消息失败表的设计
Getter
Setter
ToString
EqualsAndHashCode(of {messageId, failedPhrases})
public class MessageFailedEntity {/*** 主键*/private Long id;/*** 消息id*/private String messageId;/*** JSON格式的消息内容*/private String messageContentJsonFormat;/*** 消息类型* EMAIL 表示此消息为邮件* EMAIL_CALLBACK 表示此消息为邮件回调**/private MessageType messageType;/*** 消息失败的阶段:* PRODUCER 表示在生产者发送消息的时候失败* CONSUMER 表示在消费者消费消息的时候失败*/private MessageFailedPhrase messageFailedPhrase;/*** 失败时的异常堆栈信息*/private String failedReason;/*** 消息重试次数*/private Integer retryCount;/*** 消息重试状态* 0 表示重试失败* 1 表示重试成功*/private Integer retryStatus;/*** 时间戳*/private LocalDateTime lastUpdateTime;}重试逻辑设计
重试的思路很简单
从数据库查询消息失败表获得一批记录每次可能100条或者10000条根据实际场景自己确定根据消息类型和消息的JSON格式序列化为对应类的对象调用不同的生产者发送消息到Kafka如果该消息失败阶段是PRODUCER那么重试成功之后则更新该记录未重试成功如果该消息失败阶段是CONSUMER那么重试成功之后则只更新重试次数由对应的消费者去更新是否重试成功。因为CONSUMER只有消费成功才算重试成功。设置最大重试次数如果超过最大重试次数则不再进行重试如果是部署了多个Message Server那么执行定时重试任务时可以使用分布式锁以确保只有同一时刻只有一个Message Server在执行任务这样做的目的主要是防止为了多个任务同时进行时从数据库中查询的记录是同一批当然也可以在表中增加一个标志位来区分该记录是否在重试中来达到相同的目的根据实际情况选择即可
到此为止结合前两篇博文我们处理了在整个消息系统中可能出现所有的异常情况。
理解Rabalance
Kafka权威指南 第四章 第一节 Moving partition ownership from one consumer to another is called a rebalance. 一开始接触rebalance时候我在思考一个问题如果我的消费者还在消费消息中 此时Kafka要进行rebalance这对我的消费者业务逻辑有什么影响会不会我还在消费中然后被打断如果是这样的话那对我的消费业务逻辑的幂等性来说增加了不少挑战。
带着这些疑问我搜索了一些资料在confluent官网上发现了一篇博客详细讲了rebalance过程 文章链接
以下内容来自于上述文章链接 Suppose we have an existing consumer group with a set assignment of topic-partitions to consumers. This consumer group consists of a number of consumers, each with a member id as well as a group leader (usually the consumer that was first to join the group). A new consumer comes along and requests to join the consumer group by sending a request of JoinGroup to the Group Coordinator along with the topics it would like to subscribe to.The Group Coordinator kicks off the rebalance by telling all current members to issue their own JoinGroup requests. This is done as part of the response to the heartbeat that consumers send to the Group Coordinator to tell it they’re still alive and well.Each consumer in the group has max.poll.interval.ms to wrap up their current processing and send their JoinGroup request, at which point the world is stopped. With all of the JoinGroup requests, the Group Coordinator knows all of the consumers in the group and which topics should be part of the consumer group. It sends JoinResponses to the members, chooses a leader from amongst the members, and leaves the leader to compute the partition assignments.All group members respond with a SyncGroup request. The group leader sends its partition assignments along with its request.At this point, the Group Coordinator can send its SyncResponse to each consumer confirming their assigned topic-partitions.Finally, consumers acknowledge their assignments and processing can resume. The world is no longer stopped 第二处高亮的地方解决了我的疑问在rebalance之前会等待每个消费者把自己的消费逻辑处理完。
通过日志来理解rebalance
下面的日志是我本地的一次rebalance期间的日志可以对照上述步骤加深理解
2023-11-08T02:23:04.180-0500 kafka-coordinator-heartbeat-thread | low-priority-email-group org.apache.kafka.clients.consumer.internals.ConsumerCoordinator WARN: [Consumer instanceId2, clientIdconsumer-low-priority-email-group-2, groupIdlow-priority-email-group] consumer poll timeout has expired. This means 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 processing messages. 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.
2023-11-08T02:23:04.180-0500 kafka-coordinator-heartbeat-thread | low-priority-email-group org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId2, clientIdconsumer-low-priority-email-group-2, groupIdlow-priority-email-group] Resetting generation and member id due to: consumer pro-actively leaving the group
2023-11-08T02:23:04.180-0500 kafka-coordinator-heartbeat-thread | low-priority-email-group org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId2, clientIdconsumer-low-priority-email-group-2, groupIdlow-priority-email-group] Request joining group due to: consumer pro-actively leaving the group2023-11-08T02:32:21.607-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.620-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-4
2023-11-08T02:32:21.621-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.712-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.712-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-9
2023-11-08T02:32:21.712-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.723-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.723-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-8
2023-11-08T02:32:21.723-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.739-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.739-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-1
2023-11-08T02:32:21.739-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.741-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.741-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-5
2023-11-08T02:32:21.741-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.745-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.745-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-6
2023-11-08T02:32:21.745-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.752-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.753-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-3
2023-11-08T02:32:21.753-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.903-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.903-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-0
2023-11-08T02:32:21.903-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.919-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Request joining group due to: group is already rebalancing
2023-11-08T02:32:21.919-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Revoke previously assigned partitions low-priority-email-7
2023-11-08T02:32:21.919-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] (Re-)joining group
2023-11-08T02:32:21.920-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId1-f1f4f621-3bde-4e02-9621-a706320300ae, protocolrange}
2023-11-08T02:32:21.920-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId10-cd494fa6-1bf4-4dd8-ae28-a494b21ad247, protocolrange}
2023-11-08T02:32:21.920-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId7-e77860ca-7059-4a38-bfc9-7db3cc862a38, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId3-ec05c279-8059-4f03-b804-3da299f93b88, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId5-d08406a6-2bd1-4bed-aed9-5f65d7f75260, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId9-bab69ccd-a0b1-4f94-99d5-869fce905f3e, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId6-67b38de0-90e0-4f53-aaae-49c10a91a463, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId4-b24fc609-920a-4cc3-b507-2a4a1f1a568b, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Successfully joined group with generation Generation{generationId12, memberId8-70cd2b5d-e17f-4346-bfcb-b170e766db39, protocolrange}
2023-11-08T02:32:21.921-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Finished assignment for group at generation 12: {10-cd494fa6-1bf4-4dd8-ae28-a494b21ad247Assignment(partitions[low-priority-email-2]), 7-e77860ca-7059-4a38-bfc9-7db3cc862a38Assignment(partitions[low-priority-email-7]), 3-ec05c279-8059-4f03-b804-3da299f93b88Assignment(partitions[low-priority-email-3]), 6-67b38de0-90e0-4f53-aaae-49c10a91a463Assignment(partitions[low-priority-email-6]), 8-70cd2b5d-e17f-4346-bfcb-b170e766db39Assignment(partitions[low-priority-email-8]), 1-f1f4f621-3bde-4e02-9621-a706320300aeAssignment(partitions[low-priority-email-0, low-priority-email-1]), 4-b24fc609-920a-4cc3-b507-2a4a1f1a568bAssignment(partitions[low-priority-email-4]), 5-d08406a6-2bd1-4bed-aed9-5f65d7f75260Assignment(partitions[low-priority-email-5]), 9-bab69ccd-a0b1-4f94-99d5-869fce905f3eAssignment(partitions[low-priority-email-9])}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId10-cd494fa6-1bf4-4dd8-ae28-a494b21ad247, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId1-f1f4f621-3bde-4e02-9621-a706320300ae, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-0, low-priority-email-1])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-2])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-2
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-0, low-priority-email-1
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId3-ec05c279-8059-4f03-b804-3da299f93b88, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId5-d08406a6-2bd1-4bed-aed9-5f65d7f75260, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-3])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-3
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId7-e77860ca-7059-4a38-bfc9-7db3cc862a38, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-5])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId8-70cd2b5d-e17f-4346-bfcb-b170e766db39, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-7])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId4-b24fc609-920a-4cc3-b507-2a4a1f1a568b, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-7
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-5
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-8])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId9-bab69ccd-a0b1-4f94-99d5-869fce905f3e, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Successfully synced group in generation Generation{generationId12, memberId6-67b38de0-90e0-4f53-aaae-49c10a91a463, protocolrange}
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-4])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-8
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-9])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-4
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Notifying assignor about the new Assignment(partitions[low-priority-email-6])
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-9
2023-11-08T02:32:21.922-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Adding newly assigned partitions: low-priority-email-6
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-9 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId10, clientIdconsumer-low-priority-email-group-10, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-2 to the committed offset FetchPosition{offset13436, offsetEpochOptional[8], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch8}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-1 to the committed offset FetchPosition{offset13301, offsetEpochOptional[8], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch8}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-0 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId1, clientIdconsumer-low-priority-email-group-1, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-0 to the committed offset FetchPosition{offset24087, offsetEpochOptional[8], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch8}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-2 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId3, clientIdconsumer-low-priority-email-group-3, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-3 to the committed offset FetchPosition{offset13299, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-3 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId4, clientIdconsumer-low-priority-email-group-4, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-4 to the committed offset FetchPosition{offset13352, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-7 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId8, clientIdconsumer-low-priority-email-group-8, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-8 to the committed offset FetchPosition{offset13190, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-8 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId9, clientIdconsumer-low-priority-email-group-9, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-9 to the committed offset FetchPosition{offset13151, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-5 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId6, clientIdconsumer-low-priority-email-group-6, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-6 to the committed offset FetchPosition{offset13211, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-6 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId7, clientIdconsumer-low-priority-email-group-7, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-7 to the committed offset FetchPosition{offset13303, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}
2023-11-08T02:32:21.923-0500 consumer-low-priority-email-pool-4 org.apache.kafka.clients.consumer.internals.ConsumerCoordinator INFO: [Consumer instanceId5, clientIdconsumer-low-priority-email-group-5, groupIdlow-priority-email-group] Setting offset for partition low-priority-email-5 to the committed offset FetchPosition{offset13338, offsetEpochOptional[2], currentLeaderLeaderAndEpoch{leaderOptional[192.168.100.60:9093 (id: 1 rack: null)], epoch2}}参考资料
confluent解释rebalance的文章infoq上关于rebalance的文章
结语
至此为止利用Kafka实现一个消息系统就基本完成了所有关键的代码都在不同的博文中并进行了详细说明说过想要体会完整的设计、实现思路请移步源码仓库获取完整代码。
下一篇关于Kafak的博文打算分享一下如何利用Kafka Connect将Oracle数据库的数据同步到Postgre SQL中。
示例源码仓库
Github地址
