• RocketMQ源码阅读（六）CommitLog

    目录

    CommitLog

    load/start

    asyncPutMessage

    GroupCommitService

    waitForRunning

    onWaitEnd

    putRequest

    doCommit

    CommitRealTimeService

    FlushRealTimeService

    PutMessageLock

    CommitLog

    MappedFile是一个文件的映射，MappedFileQueue是一组文件的列表，而CommitLog类描述的是整个CommitLog文件目录

    源码4.8.0

    1.    public CommitLog(final DefaultMessageStore defaultMessageStore) {
    2.         // 文件列表
    3.         this.mappedFileQueue = new MappedFileQueue(defaultMessageStore.getMessageStoreConfig().getStorePathCommitLog(),
    4.             defaultMessageStore.getMessageStoreConfig().getMappedFileSizeCommitLog(), defaultMessageStore.getAllocateMappedFileService());
    5.         this.defaultMessageStore = defaultMessageStore;
    6.  
    7.         // 刷盘策略，同步和异步刷盘
    8.         if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
    9.             this.flushCommitLogService = new GroupCommitService();
    10.         } else {
    11.             this.flushCommitLogService = new FlushRealTimeService();
    12.         }
    13.  
    14.         // 异步刷盘时，定时将直接内存暂存区的数据提交到fileChannel，如果是同步刷盘那没啥用
    15.         this.commitLogService = new CommitRealTimeService();
    16.  
    17.         // 用来写入数据，将数据在文件末尾append
    18.         this.appendMessageCallback = new DefaultAppendMessageCallback(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
    19.         batchEncoderThreadLocal = new ThreadLocal<MessageExtBatchEncoder>() {
    20.             @Override
    21.             protected MessageExtBatchEncoder initialValue() {
    22.                 return new MessageExtBatchEncoder(defaultMessageStore.getMessageStoreConfig().getMaxMessageSize());
    23.             }
    24.         };
    25.  
    26.         // 因为是单个文件，需要考虑并发，这是写入的时候加的锁
    27.         // 有ReentrantLock和自旋两种，默认自旋
    28.         this.putMessageLock = defaultMessageStore.getMessageStoreConfig().isUseReentrantLockWhenPutMessage() ? new PutMessageReentrantLock() : new PutMessageSpinLock();
    29.     }

    load/start

    加载文件列表（将所有文件映射为MappedFile）

    启动刷盘线程，如果是异步刷盘，启动异步提交线程

    1.     public boolean load() {
    2.         boolean result = this.mappedFileQueue.load();
    3.         log.info("load commit log " + (result ? "OK" : "Failed"));
    4.         return result;
    5.     }
    6.  
    7.     public void start() {
    8.         this.flushCommitLogService.start();
    9.  
    10.         if (defaultMessageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
    11.             this.commitLogService.start();
    12.         }
    13.     }

    asyncPutMessage

    异步写入消息

    加写锁（自旋或ReentrantLock）、调用MappedFile.appendMessage在文件末尾append消息、解锁、刷盘和同步slave申请

    1.    public CompletableFuture<PutMessageResult> asyncPutMessage(final MessageExtBrokerInner msg) {
    2.         // Set the storage time
    3.         msg.setStoreTimestamp(System.currentTimeMillis());
    4.         // Set the message body BODY CRC (consider the most appropriate setting
    5.         // on the client)
    6.         msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
    7.         // Back to Results
    8.         AppendMessageResult result = null;
    9.  
    10.         StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
    11.  
    12.         String topic = msg.getTopic();
    13.         int queueId = msg.getQueueId();
    14.  
    15.         // 应该是事务消息相关的，先忽略
    16.         final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
    17.         if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
    18.                 || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
    19.             // Delay Delivery
    20.             if (msg.getDelayTimeLevel() > 0) {
    21.                 if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
    22.                     msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
    23.                 }
    24.  
    25.                 topic = TopicValidator.RMQ_SYS_SCHEDULE_TOPIC;
    26.                 queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());
    27.  
    28.                 // Backup real topic, queueId
    29.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
    30.                 MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
    31.                 msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));
    32.  
    33.                 msg.setTopic(topic);
    34.                 msg.setQueueId(queueId);
    35.             }
    36.         }
    37.  
    38.         long elapsedTimeInLock = 0;
    39.         MappedFile unlockMappedFile = null;
    40.         // 获取文件列表的最后一个文件准备写入
    41.         MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();
    42.  
    43.         // 写入锁，异步刷盘默认自旋
    44.         putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
    45.         try {
    46.             long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
    47.             this.beginTimeInLock = beginLockTimestamp;
    48.  
    49.             // Here settings are stored timestamp, in order to ensure an orderly
    50.             // global
    51.             msg.setStoreTimestamp(beginLockTimestamp);
    52.  
    53.             // 如果文件写满了，创建新文件返回
    54.             if (null == mappedFile || mappedFile.isFull()) {
    55.                 mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
    56.             }
    57.             if (null == mappedFile) {
    58.                 log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
    59.                 beginTimeInLock = 0;
    60.                 return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null));
    61.             }
    62.  
    63.             // 同步写mappedByteBuffer，异步写writeBuffer暂存区
    64.             result = mappedFile.appendMessage(msg, this.appendMessageCallback);
    65.             switch (result.getStatus()) {
    66.                 case PUT_OK:
    67.                     break;
    68.                 case END_OF_FILE:
    69.                     // 如果这个mappedFile剩余空间无法写入这条消息，会返回END_OF_FILE
    70.                     // 1、unlockMappedFile为了下边释放内存锁定
    71.                     // 2、mappedFileQueue.getLastMappedFile(0);为了创建一个新文件，然后重新写入
    72.                     unlockMappedFile = mappedFile;
    73.                     // Create a new file, re-write the message
    74.                     mappedFile = this.mappedFileQueue.getLastMappedFile(0);
    75.                     if (null == mappedFile) {
    76.                         // XXX: warn and notify me
    77.                         log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
    78.                         beginTimeInLock = 0;
    79.                         return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result));
    80.                     }
    81.                     result = mappedFile.appendMessage(msg, this.appendMessageCallback);
    82.                     break;
    83.                 case MESSAGE_SIZE_EXCEEDED:
    84.                 case PROPERTIES_SIZE_EXCEEDED:
    85.                     beginTimeInLock = 0;
    86.                     return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result));
    87.                 case UNKNOWN_ERROR:
    88.                     beginTimeInLock = 0;
    89.                     return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
    90.                 default:
    91.                     beginTimeInLock = 0;
    92.                     return CompletableFuture.completedFuture(new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result));
    93.             }
    94.  
    95.             elapsedTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
    96.             beginTimeInLock = 0;
    97.         } finally {
    98.             // 释放消息写入锁
    99.             putMessageLock.unlock();
    100.         }
    101.  
    102.         if (elapsedTimeInLock > 500) {
    103.             log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", elapsedTimeInLock, msg.getBody().length, result);
    104.         }
    105.  
    106.         // 如果WarmMapedFileEnable这个配置开启，那么MappedFile会进行内存锁定，防止在内存不足时操作系统将mappedFile部分页回收而导致使用时的缺页中断
    107.         // 这里unlock即释放这个文件的内存锁定
    108.         if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
    109.             this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
    110.         }
    111.  
    112.         PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);
    113.  
    114.         // Statistics 记录该主题的消息写入数量
    115.         storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
    116.         storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());
    117.  
    118.         // 异步刷盘MappedByteBuffer 和 暂存区+fileChannel
    119.         CompletableFuture<PutMessageStatus> flushResultFuture = submitFlushRequest(result, msg);
    120.         // 提交同步slave申请，如果是 SYNC_MASTER,马上将信息同步至SLAVE; 若ASYNC_MASTER,则每隔1S唤醒SLAVE同步请求
    121.         CompletableFuture<PutMessageStatus> replicaResultFuture = submitReplicaRequest(result, msg);
    122.         return flushResultFuture.thenCombine(replicaResultFuture, (flushStatus, replicaStatus) -> {
    123.             if (flushStatus != PutMessageStatus.PUT_OK) {
    124.                 putMessageResult.setPutMessageStatus(flushStatus);
    125.             }
    126.             if (replicaStatus != PutMessageStatus.PUT_OK) {
    127.                 putMessageResult.setPutMessageStatus(replicaStatus);
    128.                 if (replicaStatus == PutMessageStatus.FLUSH_SLAVE_TIMEOUT) {
    129.                     log.error("do sync transfer other node, wait return, but failed, topic: {} tags: {} client address: {}",
    130.                             msg.getTopic(), msg.getTags(), msg.getBornHostNameString());
    131.                 }
    132.             }
    133.             return putMessageResult;
    134.         });
    135.     }

    GroupCommitService

    同步刷盘线程，有两个列表，读写分离

    消息写入时提交刷盘申请写入到requestsWrite ，在GroupCommitService准备刷盘时将这两个列表互换，使用requestsRead来进行刷盘，requestsWrite继续接收写入请求

    1.    class GroupCommitService extends FlushCommitLogService {
    2.         private volatile List<GroupCommitRequest> requestsWrite = new ArrayList<GroupCommitRequest>();
    3.         private volatile List<GroupCommitRequest> requestsRead = new ArrayList<GroupCommitRequest>();
    4. }
    1.        public void run() {
    2.             CommitLog.log.info(this.getServiceName() + " service started");
    3.  
    4.             while (!this.isStopped()) {
    5.                 try {
    6.                     // 等待10ms，中途如果来刷盘请求，会被唤醒
    7.                     this.waitForRunning(10);
    8.  
    9.                     // 刷盘
    10.                     this.doCommit();
    11.                 } catch (Exception e) {
    12.                     CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
    13.                 }
    14.             }
    15.  
    16.             // 终止线程之前在进行一次刷盘
    17.             // Under normal circumstances shutdown, wait for the arrival of the
    18.             // request, and then flush
    19.             try {
    20.                 Thread.sleep(10);
    21.             } catch (InterruptedException e) {
    22.                 CommitLog.log.warn("GroupCommitService Exception, ", e);
    23.             }
    24.  
    25.             synchronized (this) {
    26.                 this.swapRequests();
    27.             }
    28.  
    29.             this.doCommit();
    30.  
    31.             CommitLog.log.info(this.getServiceName() + " service end");
    32.         }

    waitForRunning

    1.     protected void waitForRunning(long interval) {
    2.         // 如果来请求调用了wakeup，hasNotified标志位会设置为true
    3.         if (hasNotified.compareAndSet(true, false)) {
    4.             // wait结束后的操作
    5.             this.onWaitEnd();
    6.             return;
    7.         }
    8.  
    9.         //entry to wait
    10.         // waitPoint就是CountDownLatch，不过增加了个reset方法来重置state标志位
    11.         waitPoint.reset();
    12.  
    13.         try {
    14.             // 等待
    15.             waitPoint.await(interval, TimeUnit.MILLISECONDS);
    16.         } catch (InterruptedException e) {
    17.             log.error("Interrupted", e);
    18.         } finally {
    19.             // 等待超时或者被唤醒之后，调用onWaitEnd
    20.             hasNotified.set(false);
    21.             this.onWaitEnd();
    22.         }
    23.     }

    onWaitEnd

    互换两个列表 

    1.         protected void onWaitEnd() {
    2.             this.swapRequests();
    3.         }
    4.  
    5.         private void swapRequests() {
    6.             List<GroupCommitRequest> tmp = this.requestsWrite;
    7.             this.requestsWrite = this.requestsRead;
    8.             this.requestsRead = tmp;
    9.         }

    putRequest

    来新消息后，写入请求列表，唤醒run()方法 

    1.         public synchronized void putRequest(final GroupCommitRequest request) {
    2.             synchronized (this.requestsWrite) {
    3.                 this.requestsWrite.add(request);
    4.             }
    5.             this.wakeup();
    6.         }

    doCommit

    刷盘方法 

    1.         private void doCommit() {
    2.             synchronized (this.requestsRead) {
    3.                 if (!this.requestsRead.isEmpty()) {
    4.                     for (GroupCommitRequest req : this.requestsRead) {
    5.                         // There may be a message in the next file, so a maximum of
    6.                         // two times the flush
    7.                         // 因为这里的requestsRead可能是多条消息，并且可能跨了两个文件（第一条消息在file1末尾，第2条在file2头）
    8.                         // 所以如果FlushedWhere还没到最大的消息偏移量位置，尝试再次刷新
    9.                         boolean flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
    10.                         for (int i = 0; i < 2 && !flushOK; i++) {
    11.                             CommitLog.this.mappedFileQueue.flush(0);
    12.                             flushOK = CommitLog.this.mappedFileQueue.getFlushedWhere() >= req.getNextOffset();
    13.                         }
    14.                         // 唤醒消费者
    15.                         req.wakeupCustomer(flushOK ? PutMessageStatus.PUT_OK : PutMessageStatus.FLUSH_DISK_TIMEOUT);
    16.                     }
    17.  
    18.                     // 记录刷盘时间
    19.                     long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
    20.                     if (storeTimestamp > 0) {
    21.                         CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
    22.                     }
    23.  
    24.                     this.requestsRead.clear();
    25.                 } else {
    26.                     // Because of individual messages is set to not sync flush, it
    27.                     // will come to this process
    28.                     CommitLog.this.mappedFileQueue.flush(0);
    29.                 }
    30.             }
    31.         }

    CommitRealTimeService

    异步策略的提交线程，定时将暂存区的数据提交到fileChannel

    1.     class CommitRealTimeService extends FlushCommitLogService {
    2.  
    3.         private long lastCommitTimestamp = 0;
    4.  
    5.         @Override
    6.         public String getServiceName() {
    7.             return CommitRealTimeService.class.getSimpleName();
    8.         }
    9.  
    10.         @Override
    11.         public void run() {
    12.             CommitLog.log.info(this.getServiceName() + " service started");
    13.             while (!this.isStopped()) {
    14.                 // 每次尝试提交的间隔时间，默认200ms
    15.                 int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitIntervalCommitLog();
    16.  
    17.                 // 一次至少提交多少页，默认4
    18.                 int commitDataLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogLeastPages();
    19.  
    20.                 // 上边两个参数，如果当前只有3页，它是不提交的
    21.                 // commitDataThoroughInterval这个参数默认200，是定义了如果超过200ms没有commit，将所有数据都提交一遍
    22.                 int commitDataThoroughInterval =
    23.                     CommitLog.this.defaultMessageStore.getMessageStoreConfig().getCommitCommitLogThoroughInterval();
    24.  
    25.                 long begin = System.currentTimeMillis();
    26.                 // 判断超时
    27.                 if (begin >= (this.lastCommitTimestamp + commitDataThoroughInterval)) {
    28.                     this.lastCommitTimestamp = begin;
    29.                     commitDataLeastPages = 0;
    30.                 }
    31.  
    32.                 try {
    33.                     // 提交到fileChannel
    34.                     boolean result = CommitLog.this.mappedFileQueue.commit(commitDataLeastPages);
    35.                     long end = System.currentTimeMillis();
    36.                     if (!result) {
    37.                         // 返回false表示有数据被提交了，更新时间戳，唤醒刷盘线程
    38.                         this.lastCommitTimestamp = end; // result = false means some data committed.
    39.                         //now wake up flush thread.
    40.                         flushCommitLogService.wakeup();
    41.                     }
    42.  
    43.                     if (end - begin > 500) {
    44.                         log.info("Commit data to file costs {} ms", end - begin);
    45.                     }
    46.                     // 定时等待
    47.                     this.waitForRunning(interval);
    48.                 } catch (Throwable e) {
    49.                     CommitLog.log.error(this.getServiceName() + " service has exception. ", e);
    50.                 }
    51.             }
    52.  
    53.             // 如果线程被停止了，最后尝试提交一波
    54.             boolean result = false;
    55.             for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
    56.                 result = CommitLog.this.mappedFileQueue.commit(0);
    57.                 CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
    58.             }
    59.             CommitLog.log.info(this.getServiceName() + " service end");
    60.         }
    61.     }

    FlushRealTimeService

    异步策略的刷盘线程，代码格式类似CommitRealTimeService

    1.     class FlushRealTimeService extends FlushCommitLogService {
    2.         private long lastFlushTimestamp = 0;
    3.         private long printTimes = 0;
    4.  
    5.         public void run() {
    6.             CommitLog.log.info(this.getServiceName() + " service started");
    7.  
    8.             while (!this.isStopped()) {
    9.                 // true表示定时刷盘，false也是定时，但也会接收commit的唤醒，默认false
    10.                 boolean flushCommitLogTimed = CommitLog.this.defaultMessageStore.getMessageStoreConfig().isFlushCommitLogTimed();
    11.  
    12.                 // 间隔时间，默认500ms
    13.                 int interval = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushIntervalCommitLog();
    14.                 // 最少刷多少个页，默认4
    15.                 int flushPhysicQueueLeastPages = CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogLeastPages();
    16.  
    17.                 // 和commit一样，这个参数表示，超时之后刷所有
    18.                 int flushPhysicQueueThoroughInterval =
    19.                     CommitLog.this.defaultMessageStore.getMessageStoreConfig().getFlushCommitLogThoroughInterval();
    20.  
    21.                 boolean printFlushProgress = false;
    22.  
    23.                 // Print flush progress
    24.                 long currentTimeMillis = System.currentTimeMillis();
    25.                 if (currentTimeMillis >= (this.lastFlushTimestamp + flushPhysicQueueThoroughInterval)) {
    26.                     this.lastFlushTimestamp = currentTimeMillis;
    27.                     flushPhysicQueueLeastPages = 0;
    28.                     printFlushProgress = (printTimes++ % 10) == 0;
    29.                 }
    30.  
    31.                 try {
    32.                     // 定时等待，waitForRunning会接收commit线程的唤醒
    33.                     if (flushCommitLogTimed) {
    34.                         Thread.sleep(interval);
    35.                     } else {
    36.                         this.waitForRunning(interval);
    37.                     }
    38.                     // 定时打印 忽略
    39.                     if (printFlushProgress) {
    40.                         this.printFlushProgress();
    41.                     }
    42.                     // 刷盘
    43.                     long begin = System.currentTimeMillis();
    44.                     CommitLog.this.mappedFileQueue.flush(flushPhysicQueueLeastPages);
    45.                     long storeTimestamp = CommitLog.this.mappedFileQueue.getStoreTimestamp();
    46.                     if (storeTimestamp > 0) {
    47.                         CommitLog.this.defaultMessageStore.getStoreCheckpoint().setPhysicMsgTimestamp(storeTimestamp);
    48.                     }
    49.                     long past = System.currentTimeMillis() - begin;
    50.                     if (past > 500) {
    51.                         log.info("Flush data to disk costs {} ms", past);
    52.                     }
    53.                 } catch (Throwable e) {
    54.                     CommitLog.log.warn(this.getServiceName() + " service has exception. ", e);
    55.                     this.printFlushProgress();
    56.                 }
    57.             }
    58.  
    59.             // shutdown之后的再次尝试刷盘
    60.             // Normal shutdown, to ensure that all the flush before exit
    61.             boolean result = false;
    62.             for (int i = 0; i < RETRY_TIMES_OVER && !result; i++) {
    63.                 result = CommitLog.this.mappedFileQueue.flush(0);
    64.                 CommitLog.log.info(this.getServiceName() + " service shutdown, retry " + (i + 1) + " times " + (result ? "OK" : "Not OK"));
    65.             }
    66.  
    67.             this.printFlushProgress();
    68.  
    69.             CommitLog.log.info(this.getServiceName() + " service end");
    70.         }

    PutMessageLock

    往mappedFile写数据时的写入锁，两种实现，ReentrantLock和do..while循环自旋锁，默认自旋，因为默认是异步刷盘，冲突不大，所以自旋性能更好

    1. public class PutMessageReentrantLock implements PutMessageLock {
    2.     private ReentrantLock putMessageNormalLock = new ReentrantLock(); // NonfairSync
    3.  
    4.     @Override
    5.     public void lock() {
    6.         putMessageNormalLock.lock();
    7.     }
    8.  
    9.     @Override
    10.     public void unlock() {
    11.         putMessageNormalLock.unlock();
    12.     }
    13. }
    14.  
    15. // 自旋，do..while..循环，cas操作
    16. public class PutMessageSpinLock implements PutMessageLock {
    17.     //true: Can lock, false : in lock.
    18.     private AtomicBoolean putMessageSpinLock = new AtomicBoolean(true);
    19.  
    20.     @Override
    21.     public void lock() {
    22.         boolean flag;
    23.         do {
    24.             flag = this.putMessageSpinLock.compareAndSet(true, false);
    25.         }
    26.         while (!flag);
    27.     }
    28.  
    29.     @Override
    30.     public void unlock() {
    31.         this.putMessageSpinLock.compareAndSet(false, true);
    32.     }
    33. }

  • 相关阅读:
    MySQL唯一约束（UNIQUE KEY）
    postgreSQL中将字段从smallint类型改成boolean类型
    在华为云服务器上CentOS 7安装单机版Redis
    聊聊分布式架构02——Http到Https
    kafka权限认证 topic权限认证 权限动态认证-亲测成功
    Java 8 中Stream流的一些用法
    Mysql应用日志时间与系统时间相差八小时
    python学习05协程_2
    【全开源】JAVA打车小程序APP打车顺风车滴滴车跑腿源码微信小程序打车源码
    【python】OpenCV—Color Detection
  • 原文地址：https://blog.csdn.net/xyjy11/article/details/125398699