KafkaProducer Sender 线程详解(含详细的执行流程图)

Stella981
• 阅读 711

温馨提示:本文基于 Kafka 2.2.1 版本。如果觉得源码阅读比较枯燥,本文的中间有 Sender 线程的工作流程图。

上文 《源码分析 Kafka 消息发送流程》 已经详细介绍了 KafkaProducer send 方法的流程,该方法只是将消息追加到 KafKaProducer 的缓存中,并未真正的向 broker 发送消息,本文将来探讨 Kafka 的 Sender 线程。

在 KafkaProducer 中会启动一个单独的线程,其名称为 “kafka-producer-network-thread | clientID”,其中 clientID 为生产者的 id 。

1、Sender 线程详解

1.1 类图

KafkaProducer Sender 线程详解(含详细的执行流程图)

我们先来看一下其各个属性的含义:

  • KafkaClient client
    kafka 网络通信客户端,主要封装与 broker 的网络通信。

  • RecordAccumulator accumulator
    消息记录累积器,消息追加的入口(RecordAccumulator 的 append 方法)。

  • Metadata metadata
    元数据管理器,即 topic 的路由分区信息。

  • boolean guaranteeMessageOrder
    是否需要保证消息的顺序性。

  • int maxRequestSize
    调用 send 方法发送的最大请求大小,包括 key、消息体序列化后的消息总大小不能超过该值。通过参数 max.request.size 来设置。

  • short acks
    用来定义消息“已提交”的条件(标准),就是 Broker 端向客户端承偌已提交的条件,可选值如下0、-1、1.

  • int retries
    重试次数。

  • Time time
    时间工具类。

  • boolean running
    该线程状态,为 true 表示运行中。

  • boolean forceClose
    是否强制关闭,此时会忽略正在发送中的消息。

  • SenderMetrics sensors
    消息发送相关的统计指标收集器。

  • int requestTimeoutMs
    请求的超时时间。

  • long retryBackoffMs
    请求失败时在重试之前等待的时间。

  • ApiVersions apiVersions
    API版本信息。

  • TransactionManager transactionManager
    事务处理器。

  • Map< TopicPartition, List< ProducerBatch>> inFlightBatches
    正在执行发送相关的消息批次。

1.2 run 方法详解

Sender#run

public void run() {    log.debug("Starting Kafka producer I/O thread.");    while (running) {           try {            runOnce();    // @1        } catch (Exception e) {            log.error("Uncaught error in kafka producer I/O thread: ", e);        }    }    log.debug("Beginning shutdown of Kafka producer I/O thread, sending remaining records.");    while (!forceClose && (this.accumulator.hasUndrained() || this.client.inFlightRequestCount() > 0)) {    // @2        try {            runOnce();        } catch (Exception e) {            log.error("Uncaught error in kafka producer I/O thread: ", e);        }    }    if (forceClose) {                                                                                                                                     // @3        log.debug("Aborting incomplete batches due to forced shutdown");        this.accumulator.abortIncompleteBatches();    }    try {        this.client.close();                                                                                                                               // @4    } catch (Exception e) {        log.error("Failed to close network client", e);    }    log.debug("Shutdown of Kafka producer I/O thread has completed.");}

代码@1:Sender 线程在运行状态下主要的业务处理方法,将消息缓存区中的消息向 broker 发送。

代码@2:如果主动关闭 Sender 线程,如果不是强制关闭,则如果缓存区还有消息待发送,再次调用 runOnce 方法将剩余的消息发送完毕后再退出。

代码@3:如果强制关闭 Sender 线程,则拒绝未完成提交的消息。

代码@4:关闭 Kafka Client 即网络通信对象。

接下来将分别探讨其上述方法的实现细节。

1.2.1 runOnce 详解

Sender#runOnce

void runOnce() {    // 此处省略与事务消息相关的逻辑    long currentTimeMs = time.milliseconds();    long pollTimeout = sendProducerData(currentTimeMs);   // @1    client.poll(pollTimeout, currentTimeMs);                            // @2}

本文不关注事务消息的实现原理,故省略了该部分的代码。

代码@1:调用 sendProducerData 方法发送消息。
代码@2:调用这个方法的作用?

接下来分别对上述两个方法进行深入探究。

1.1.2.1 sendProducerData

接下来将详细分析其实现步骤。
Sender#sendProducerData

Cluster cluster = metadata.fetch();// get the list of partitions with data ready to sendRecordAccumulator.ReadyCheckResult result = this.accumulator.ready(cluster, now);

Step1:首先根据当前时间,根据缓存队列中的数据判断哪些 topic 的 哪些分区已经达到发送条件。达到可发送的条件将在 2.1.1.1 节详细分析。

Sender#sendProducerData

if (!result.unknownLeaderTopics.isEmpty()) {    for (String topic : result.unknownLeaderTopics)        this.metadata.add(topic);    log.debug("Requesting metadata update due to unknown leader topics from the batched records: {}",                result.unknownLeaderTopics);    this.metadata.requestUpdate();}

Step2:如果在待发送的消息未找到其路由信息,则需要首先去 broker 服务器拉取对应的路由信息(分区的 leader 节点信息)。

Sender#sendProducerData

long notReadyTimeout = Long.MAX_VALUE;while (iter.hasNext()) {    Node node = iter.next();    if (!this.client.ready(node, now)) {        iter.remove();        notReadyTimeout = Math.min(notReadyTimeout, this.client.pollDelayMs(node, now));    }}

Step3:移除在网络层面没有准备好的分区,并且计算在接下来多久的时间间隔内,该分区都将处于未准备状态。

1、在网络环节没有准备好的标准如下:

  • 分区没有未完成的更新元素数据请求(metadata)。

  • 当前生产者与对端 broker 已建立连接并完成了 TCP 的三次握手。

  • 如果启用 SSL、ACL 等机制,相关状态都已就绪。

  • 该分区对应的连接正在处理中的请求数时是否超过设定值,默认为 5,可通过属性 max.in.flight.requests.per.connection 来设置。

2、client pollDelayMs 预估分区在接下来多久的时间间隔内都将处于未转变好状态(not ready),其标准如下:

  • 如果已与对端的 TCP 连接已创建好,并处于已连接状态,此时如果没有触发限流,则返回0,如果有触发限流,则返回限流等待时间。

  • 如果还位于对端建立 TCP 连接,则返回 Long.MAX_VALUE,因为连接建立好后,会唤醒发送线程的。

Sender#sendProducerData

// create produce requestsMap<Integer, List<ProducerBatch>> batches = this.accumulator.drain(cluster, result.readyNodes, this.maxRequestSize, now);

Step4:根据已准备的分区,从缓存区中抽取待发送的消息批次( ProducerBatch ),并且按照 nodeId:List 组织,注意,抽取后的 ProducerBatch 将不能再追加消息了,就算还有剩余空间可用,具体抽取将在下文在详细介绍。

Sender#sendProducerData

addToInflightBatches(batches);public void addToInflightBatches(Map<Integer, List<ProducerBatch>> batches) {    for (List<ProducerBatch> batchList : batches.values()) {        addToInflightBatches(batchList);    }}private void addToInflightBatches(List<ProducerBatch> batches) {    for (ProducerBatch batch : batches) {        List<ProducerBatch> inflightBatchList = inFlightBatches.get(batch.topicPartition);        if (inflightBatchList == null) {            inflightBatchList = new ArrayList<>();            inFlightBatches.put(batch.topicPartition, inflightBatchList);        }        inflightBatchList.add(batch);    }}

Step5:将抽取的 ProducerBatch 加入到 inFlightBatches 数据结构,该属性的声明如下:Map<TopicPartition, List< ProducerBatch >> inFlightBatches,即按照 topic-分区 为键,存放已抽取的 ProducerBatch,这个属性的含义就是存储待发送的消息批次。可以根据该数据结构得知在消息发送时以分区为维度反馈 Sender 线程的“积压情况”,max.in.flight.requests.per.connection 就是来控制积压的最大数量,如果积压达到这个数值,针对该队列的消息发送会限流。

Sender#sendProducerData

accumulator.resetNextBatchExpiryTime();List<ProducerBatch> expiredInflightBatches = getExpiredInflightBatches(now);List<ProducerBatch> expiredBatches = this.accumulator.expiredBatches(now);expiredBatches.addAll(expiredInflightBatches);

Step6:从 inflightBatches 与 batches 中查找已过期的消息批次( ProducerBatch ),判断是否过期的标准是系统当前时间与 ProducerBatch 创建时间之差是否超过120s,过期时间可以通过参数 delivery.timeout.ms 设置。

Sender#sendProducerData

if (!expiredBatches.isEmpty())    log.trace("Expired {} batches in accumulator", expiredBatches.size());for (ProducerBatch expiredBatch : expiredBatches) {    String errorMessage = "Expiring " + expiredBatch.recordCount + " record(s) for " + expiredBatch.topicPartition                + ":" + (now - expiredBatch.createdMs) + " ms has passed since batch creation";    failBatch(expiredBatch, -1, NO_TIMESTAMP, new TimeoutException(errorMessage), false);    if (transactionManager != null && expiredBatch.inRetry()) {        // This ensures that no new batches are drained until the current in flight batches are fully resolved.        transactionManager.markSequenceUnresolved(expiredBatch.topicPartition);    }}

Step7:处理已超时的消息批次,通知该批消息发送失败,即通过设置  KafkaProducer#send 方法返回的凭证中的 FutureRecordMetadata 中的 ProduceRequestResult result,使之调用其 get 方法不会阻塞。

Sender#sendProducerData

sensors.updateProduceRequestMetrics(batches);

Step8:收集统计指标,本文不打算详细分析,但后续会专门对 Kafka 的 Metrics 设计进行一个深入的探讨与学习。

Sender#sendProducerData

long pollTimeout = Math.min(result.nextReadyCheckDelayMs, notReadyTimeout);pollTimeout = Math.min(pollTimeout, this.accumulator.nextExpiryTimeMs() - now);pollTimeout = Math.max(pollTimeout, 0);if (!result.readyNodes.isEmpty()) {    log.trace("Nodes with data ready to send: {}", result.readyNodes);    pollTimeout = 0;}

Step9:设置下一次的发送延时,待补充详细分析。

Sender#sendProducerData

sendProduceRequests(batches, now);private void sendProduceRequests(Map<Integer, List<ProducerBatch>> collated, long now) {    for (Map.Entry<Integer, List<ProducerBatch>> entry : collated.entrySet())        sendProduceRequest(now, entry.getKey(), acks, requestTimeoutMs, entry.getValue());}

Step10:该步骤按照 brokerId 分别构建发送请求,即每一个 broker 会将多个  ProducerBatch 一起封装成一个请求进行发送,同一时间,每一个 与 broker 连接只会只能发送一个请求,注意,这里只是构建请求,并最终会通过 NetworkClient#send 方法,将该批数据设置到 NetworkClient 的待发送数据中,此时并没有触发真正的网络调用。

sendProducerData 方法就介绍到这里了,既然这里还没有进行真正的网络请求,那在什么时候触发呢?

我们继续回到 runOnce 方法。

1.2.1.2 NetworkClient 的 poll 方法
 public List<ClientResponse> poll(long timeout, long now) {    ensureActive();    if (!abortedSends.isEmpty()) {        // If there are aborted sends because of unsupported version exceptions or disconnects,        // handle them immediately without waiting for Selector#poll.        List<ClientResponse> responses = new ArrayList<>();        handleAbortedSends(responses);        completeResponses(responses);        return responses;    }    long metadataTimeout = metadataUpdater.maybeUpdate(now);   // @1    try {        this.selector.poll(Utils.min(timeout, metadataTimeout, defaultRequestTimeoutMs));    // @2    } catch (IOException e) {        log.error("Unexpected error during I/O", e);    }    // process completed actions    long updatedNow = this.time.milliseconds();    List<ClientResponse> responses = new ArrayList<>();            // @3    handleCompletedSends(responses, updatedNow);    handleCompletedReceives(responses, updatedNow);    handleDisconnections(responses, updatedNow);    handleConnections();    handleInitiateApiVersionRequests(updatedNow);    handleTimedOutRequests(responses, updatedNow);    completeResponses(responses);                                               // @4    return responses;}

本文并不会详细深入探讨其网络实现部分,Kafka 的 网络通讯后续我会专门详细的介绍,在这里先点出其关键点。

代码@1:尝试更新云数据。

代码@2:触发真正的网络通讯,该方法中会通过收到调用 NIO 中的 Selector#select() 方法,对通道的读写就绪事件进行处理,当写事件就绪后,就会将通道中的消息发送到远端的 broker。

代码@3:然后会消息发送,消息接收、断开连接、API版本,超时等结果进行收集。

代码@4:并依次对结果进行唤醒,此时会将响应结果设置到  KafkaProducer#send 方法返回的凭证中,从而唤醒发送客户端,完成一次完整的消息发送流程。

Sender 发送线程的流程就介绍到这里了,接下来首先给出一张流程图,然后对上述流程中一些关键的方法再补充深入探讨一下。

1.2.2 run 方法流程图

KafkaProducer Sender 线程详解(含详细的执行流程图)

根据上面的源码分析得出上述流程图,图中对重点步骤也详细标注了其关键点。下面我们对上述流程图中 Sender 线程依赖的相关类的核心方法进行解读,以便加深 Sender 线程的理解。

由于在讲解 Sender 发送流程中,大部分都是调用 RecordAccumulator 方法来实现其特定逻辑,故接下来重点对上述涉及到 RecordAccumulator 的方法进行一个详细剖析,加强对 Sender 流程的理解。

2、RecordAccumulator 核心方法详解

2.1 RecordAccumulator 的 ready 方法详解

该方法主要就是根据缓存区中的消息,判断哪些分区已经达到发送条件。

RecordAccumulator#ready

public ReadyCheckResult ready(Cluster cluster, long nowMs) {    Set<Node> readyNodes = new HashSet<>();    long nextReadyCheckDelayMs = Long.MAX_VALUE;    Set<String> unknownLeaderTopics = new HashSet<>();    boolean exhausted = this.free.queued() > 0;    for (Map.Entry<TopicPartition, Deque<ProducerBatch>> entry : this.batches.entrySet()) {   // @1        TopicPartition part = entry.getKey();        Deque<ProducerBatch> deque = entry.getValue();        Node leader = cluster.leaderFor(part);   // @2        synchronized (deque) {            if (leader == null && !deque.isEmpty()) {   // @3                // This is a partition for which leader is not known, but messages are available to send.                // Note that entries are currently not removed from batches when deque is empty.                unknownLeaderTopics.add(part.topic());            } else if (!readyNodes.contains(leader) && !isMuted(part, nowMs)) {    // @4                ProducerBatch batch = deque.peekFirst();                if (batch != null) {                    long waitedTimeMs = batch.waitedTimeMs(nowMs);                    boolean backingOff = batch.attempts() > 0 && waitedTimeMs < retryBackoffMs;                    long timeToWaitMs = backingOff ? retryBackoffMs : lingerMs;                    boolean full = deque.size() > 1 || batch.isFull();                    boolean expired = waitedTimeMs >= timeToWaitMs;                    boolean sendable = full || expired || exhausted || closed || flushInProgress();                    if (sendable && !backingOff) {   // @5                        readyNodes.add(leader);                    } else {                        long timeLeftMs = Math.max(timeToWaitMs - waitedTimeMs, 0);                        // Note that this results in a conservative estimate since an un-sendable partition may have                        // a leader that will later be found to have sendable data. However, this is good enough                        // since we'll just wake up and then sleep again for the remaining time.                        nextReadyCheckDelayMs = Math.min(timeLeftMs, nextReadyCheckDelayMs);                       }                }            }        }    }    return new ReadyCheckResult(readyNodes, nextReadyCheckDelayMs, unknownLeaderTopics);}

代码@1:对生产者缓存区 ConcurrentHashMap<TopicPartition, Deque< ProducerBatch>> batches 遍历,从中挑选已准备好的消息批次。

代码@2:从生产者元数据缓存中尝试查找分区(TopicPartition) 的 leader 信息,如果不存在,当将该 topic 添加到 unknownLeaderTopics (代码@3),稍后会发送元数据更新请求去 broker 端查找分区的路由信息。

代码@4:如果不在 readyNodes 中就需要判断是否满足条件,isMuted 与顺序消息有关,本文暂时不关注,在后面的顺序消息部分会重点探讨。

代码@5:这里就是判断是否准备好的条件,先一个一个来解读局部变量的含义。

  • long waitedTimeMs
    该 ProducerBatch 已等待的时长,等于当前时间戳 与 ProducerBatch 的 lastAttemptMs 之差,在 ProducerBatch 创建时或需要重试时会将当前的时间赋值给lastAttemptMs。

  • retryBackoffMs
    当发生异常时发起重试之前的等待时间,默认为 100ms,可通过属性 retry.backoff.ms 配置。

  • batch.attempts()
    该批次当前已重试的次数。

  • backingOff
    后台发送是否关闭,即如果需要重试并且等待时间小于 retryBackoffMs ,则 backingOff = true,也意味着该批次未准备好。

  • timeToWaitMs
    send 线程发送消息需要的等待时间,如果 backingOff  为 true,表示该批次是在重试,并且等待时间小于系统设置的需要等待时间,这种情况下 timeToWaitMs = retryBackoffMs 。否则需要等待的时间为 lingerMs。

  • boolean full 该批次是否已满,如果两个条件中的任意一个满足即为 true。

  • Deque< ProducerBatch> 该队列的个数大于1,表示肯定有一个 ProducerBatch 已写满。

  • ProducerBatch 已写满。

  • boolean expired
    是否过期,等于已经等待的时间是否大于需要等待的时间,如果把发送看成定时发送的话,expired 为 true 表示定时器已到达触发点,即需要执行。

  • boolean exhausted
    当前生产者缓存已不够,创建新的 ProducerBatch 时阻塞在申请缓存空间的线程大于0,此时应立即将缓存区中的消息立即发送到服务器。

  • boolean sendable 是否可发送。其满足下面的任意一个条件即可:

  • 该批次已写满。(full = true)。

  • 已等待系统规定的时长。(expired = true)

  • 发送者内部缓存区已耗尽并且有新的线程需要申请(exhausted = true)。

  • 该发送者的 close 方法被调用(close = true)。

  • 该发送者的 flush 方法被调用。

2.2 RecordAccumulator 的 drain方法详解

RecordAccumulator#drain

public Map<Integer, List<ProducerBatch>> drain(Cluster cluster, Set<Node> nodes, int maxSize, long now) { // @1    if (nodes.isEmpty())        return Collections.emptyMap();    Map<Integer, List<ProducerBatch>> batches = new HashMap<>();    for (Node node : nodes) {                                                                                                                                      List<ProducerBatch> ready = drainBatchesForOneNode(cluster, node, maxSize, now);                      // @2        batches.put(node.id(), ready);    }    return batches;}

代码@1:我们首先来介绍该方法的参数:

  • Cluster cluster
    集群信息。

  • Set< Node> nodes
    已准备好的节点集合。

  • int maxSize
    一次请求最大的字节数。

  • long now
    当前时间。

代码@2:遍历所有节点,调用 drainBatchesForOneNode 方法抽取数据,组装成 Map<Integer /** brokerId */, List< ProducerBatch>> batches。

接下来重点来看一下 drainBatchesForOneNode。
RecordAccumulator#drainBatchesForOneNode

private List<ProducerBatch> drainBatchesForOneNode(Cluster cluster, Node node, int maxSize, long now) {    int size = 0;    List<PartitionInfo> parts = cluster.partitionsForNode(node.id());   // @1    List<ProducerBatch> ready = new ArrayList<>();    int start = drainIndex = drainIndex % parts.size();                        // @2    do {                                                                                                // @3         PartitionInfo part = parts.get(drainIndex);        TopicPartition tp = new TopicPartition(part.topic(), part.partition());         this.drainIndex = (this.drainIndex + 1) % parts.size();                             if (isMuted(tp, now))            continue;        Deque<ProducerBatch> deque = getDeque(tp);                              // @4        if (deque == null)            continue;        synchronized (deque) {            // invariant: !isMuted(tp,now) && deque != null            ProducerBatch first = deque.peekFirst();                                         // @5            if (first == null)                continue;            // first != null            boolean backoff = first.attempts() > 0 && first.waitedTimeMs(now) < retryBackoffMs;   // @6            // Only drain the batch if it is not during backoff period.            if (backoff)                                                                                                     continue;            if (size + first.estimatedSizeInBytes() > maxSize && !ready.isEmpty()) {     // @7                break;            } else {                if (shouldStopDrainBatchesForPartition(first, tp))                                                      break;                // 这里省略与事务消息相关的代码,后续会重点学习。                batch.close();                                                                                            // @8                size += batch.records().sizeInBytes();                ready.add(batch);                                                                                            batch.drained(now);                                                                                         }        }    } while (start != drainIndex);    return ready;}

代码@1:根据 brokerId 获取该 broker 上的所有主分区。
代码@2:初始化 start。这里首先来阐述一下 start 与 drainIndex 。

  • start 当前开始遍历的分区序号。

  • drainIndex 上次抽取的队列索引后,这里主要是为了每个队列都是从零号分区开始抽取。

代码@3:循环从缓存区抽取对应分区中累积的数据。

代码@4:根据 topic + 分区号从生产者发送缓存区中获取已累积的双端Queue。

代码@5:从双端队列的头部获取一个元素。(消息追加时是追加到队列尾部)。

代码@6:如果当前批次是重试,并且还未到阻塞时间,则跳过该分区。

代码@7:如果当前已抽取的消息总大小 加上新的消息已超过 maxRequestSize,则结束抽取。

代码@8:将当前批次加入到已准备集合中,并关闭该批次,即不在允许向该批次中追加消息。

关于消息发送就介绍到这里,NetworkClient 的 poll 方法内部会调用 Selector 执行就绪事件的选择,并将抽取的消息通过网络发送到 Broker 服务器,关于网络后面的具体实现,将在后续文章中单独介绍。

如果文章对您有所帮助的话,麻烦帮忙点【在看】,谢谢您的认可与支持。


中间件兴趣圈已经陆续发表了源码分析Dubbo、ElasticJob、Mybatis、RocketMQ系列,RocketMQ实战与案例分析、ElasticSearch使用指南等。

KafkaProducer Sender 线程详解(含详细的执行流程图)

本文分享自微信公众号 - 中间件兴趣圈(dingwpmz_zjj)。
如有侵权,请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”,欢迎正在阅读的你也加入,一起分享。

点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
待兔 待兔
3个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Jacquelyn38 Jacquelyn38
3年前
2020年前端实用代码段,为你的工作保驾护航
有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )
Stella981 Stella981
3年前
Android So动态加载 优雅实现与原理分析
背景:漫品Android客户端集成适配转换功能(基于目标识别(So库35M)和人脸识别库(5M)),导致apk体积50M左右,为优化客户端体验,决定实现So文件动态加载.!(https://oscimg.oschina.net/oscnet/00d1ff90e4b34869664fef59e3ec3fdd20b.png)点击上方“蓝字”关注我
Wesley13 Wesley13
3年前
00:Java简单了解
浅谈Java之概述Java是SUN(StanfordUniversityNetwork),斯坦福大学网络公司)1995年推出的一门高级编程语言。Java是一种面向Internet的编程语言。随着Java技术在web方面的不断成熟,已经成为Web应用程序的首选开发语言。Java是简单易学,完全面向对象,安全可靠,与平台无关的编程语言。
Stella981 Stella981
3年前
Django中Admin中的一些参数配置
设置在列表中显示的字段,id为django模型默认的主键list_display('id','name','sex','profession','email','qq','phone','status','create_time')设置在列表可编辑字段list_editable
Wesley13 Wesley13
3年前
35岁是技术人的天花板吗?
35岁是技术人的天花板吗?我非常不认同“35岁现象”,人类没有那么脆弱,人类的智力不会说是35岁之后就停止发展,更不是说35岁之后就没有机会了。马云35岁还在教书,任正非35岁还在工厂上班。为什么技术人员到35岁就应该退役了呢?所以35岁根本就不是一个问题,我今年已经37岁了,我发现我才刚刚找到自己的节奏,刚刚上路。
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
9个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这