Nicksxs's Blog

What hurts more, the pain of hard work or the pain of regret?

0%

CommitLog 结构

CommitLog 是 rocketmq 的服务端,也就是 broker 存储消息的的文件,跟 kafka 一样,也是顺序写入,当然消息是变长的,生成的规则是每个文件的默认1G =1024 * 1024 * 1024,commitlog的文件名fileName,名字长度为20位,左边补零,剩余为起始偏移量;比如00000000000000000000代表了第一个文件,起始偏移量为0,文件大小为1G=1 073 741 824Byte;当这个文件满了,第二个文件名字为00000000001073741824,起始偏移量为1073741824, 消息存储的时候会顺序写入文件,当文件满了则写入下一个文件,代码中的定义

// CommitLog file size,default is 1G
private int mapedFileSizeCommitLog = 1024 * 1024 * 1024;

kLahwW

本地跑个 demo 验证下,也是这样,这里奇妙有几个比较巧妙的点(个人观点),首先文件就刚好是 1G,并且按照大小偏移量去生成下一个文件,这样获取消息的时候按大小算一下就知道在哪个文件里了,

代码中写入 CommitLog 的逻辑可以从这开始看

public PutMessageResult putMessage(final MessageExtBrokerInner msg) {
        // Set the storage time
        msg.setStoreTimestamp(System.currentTimeMillis());
        // Set the message body BODY CRC (consider the most appropriate setting
        // on the client)
        msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
        // Back to Results
        AppendMessageResult result = null;

        StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();

        String topic = msg.getTopic();
        int queueId = msg.getQueueId();

        final int tranType = MessageSysFlag.getTransactionValue(msg.getSysFlag());
        if (tranType == MessageSysFlag.TRANSACTION_NOT_TYPE
            || tranType == MessageSysFlag.TRANSACTION_COMMIT_TYPE) {
            // Delay Delivery
            if (msg.getDelayTimeLevel() > 0) {
                if (msg.getDelayTimeLevel() > this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel()) {
                    msg.setDelayTimeLevel(this.defaultMessageStore.getScheduleMessageService().getMaxDelayLevel());
                }

                topic = ScheduleMessageService.SCHEDULE_TOPIC;
                queueId = ScheduleMessageService.delayLevel2QueueId(msg.getDelayTimeLevel());

                // Backup real topic, queueId
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_TOPIC, msg.getTopic());
                MessageAccessor.putProperty(msg, MessageConst.PROPERTY_REAL_QUEUE_ID, String.valueOf(msg.getQueueId()));
                msg.setPropertiesString(MessageDecoder.messageProperties2String(msg.getProperties()));

                msg.setTopic(topic);
                msg.setQueueId(queueId);
            }
        }

        long eclipseTimeInLock = 0;
        MappedFile unlockMappedFile = null;
        MappedFile mappedFile = this.mappedFileQueue.getLastMappedFile();

        putMessageLock.lock(); //spin or ReentrantLock ,depending on store config
        try {
            long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
            this.beginTimeInLock = beginLockTimestamp;

            // Here settings are stored timestamp, in order to ensure an orderly
            // global
            msg.setStoreTimestamp(beginLockTimestamp);

            if (null == mappedFile || mappedFile.isFull()) {
                mappedFile = this.mappedFileQueue.getLastMappedFile(0); // Mark: NewFile may be cause noise
            }
            if (null == mappedFile) {
                log.error("create mapped file1 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                beginTimeInLock = 0;
                return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, null);
            }

            result = mappedFile.appendMessage(msg, this.appendMessageCallback);
            switch (result.getStatus()) {
                case PUT_OK:
                    break;
                case END_OF_FILE:
                    unlockMappedFile = mappedFile;
                    // Create a new file, re-write the message
                    mappedFile = this.mappedFileQueue.getLastMappedFile(0);
                    if (null == mappedFile) {
                        // XXX: warn and notify me
                        log.error("create mapped file2 error, topic: " + msg.getTopic() + " clientAddr: " + msg.getBornHostString());
                        beginTimeInLock = 0;
                        return new PutMessageResult(PutMessageStatus.CREATE_MAPEDFILE_FAILED, result);
                    }
                    result = mappedFile.appendMessage(msg, this.appendMessageCallback);
                    break;
                case MESSAGE_SIZE_EXCEEDED:
                case PROPERTIES_SIZE_EXCEEDED:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.MESSAGE_ILLEGAL, result);
                case UNKNOWN_ERROR:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
                default:
                    beginTimeInLock = 0;
                    return new PutMessageResult(PutMessageStatus.UNKNOWN_ERROR, result);
            }

            eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
            beginTimeInLock = 0;
        } finally {
            putMessageLock.unlock();
        }

        if (eclipseTimeInLock > 500) {
            log.warn("[NOTIFYME]putMessage in lock cost time(ms)={}, bodyLength={} AppendMessageResult={}", eclipseTimeInLock, msg.getBody().length, result);
        }

        if (null != unlockMappedFile && this.defaultMessageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
            this.defaultMessageStore.unlockMappedFile(unlockMappedFile);
        }

        PutMessageResult putMessageResult = new PutMessageResult(PutMessageStatus.PUT_OK, result);

        // Statistics
        storeStatsService.getSinglePutMessageTopicTimesTotal(msg.getTopic()).incrementAndGet();
        storeStatsService.getSinglePutMessageTopicSizeTotal(topic).addAndGet(result.getWroteBytes());

        handleDiskFlush(result, putMessageResult, msg);
        handleHA(result, putMessageResult, msg);

        return putMessageResult;
    }

前面也看到在CommitLog 目录下是有大小为 1G 的文件组成,在实现逻辑中,其实是通过 org.apache.rocketmq.store.MappedFileQueue ,内部是存的一个MappedFile的队列,对于写入的场景每次都是通过org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile() 获取最后一个文件,如果还没有创建,或者最后这个文件已经满了,那就调用 org.apache.rocketmq.store.MappedFileQueue#getLastMappedFile(long)

public MappedFile getLastMappedFile(final long startOffset, boolean needCreate) {
        long createOffset = -1;
  			// 调用前面的方法,只是从 mappedFileQueue 获取最后一个
        MappedFile mappedFileLast = getLastMappedFile();

        // 如果为空,计算下创建的偏移量
        if (mappedFileLast == null) {
            createOffset = startOffset - (startOffset % this.mappedFileSize);
        }
  
				// 如果不为空,但是当前的文件写满了
        if (mappedFileLast != null && mappedFileLast.isFull()) {
            // 前一个的偏移量加上单个文件的偏移量,也就是 1G
            createOffset = mappedFileLast.getFileFromOffset() + this.mappedFileSize;
        }

        if (createOffset != -1 && needCreate) {
            // 根据 createOffset 转换成文件名进行创建
            String nextFilePath = this.storePath + File.separator + UtilAll.offset2FileName(createOffset);
            String nextNextFilePath = this.storePath + File.separator
                + UtilAll.offset2FileName(createOffset + this.mappedFileSize);
            MappedFile mappedFile = null;

          	// 这里如果allocateMappedFileService 存在,就提交请求
            if (this.allocateMappedFileService != null) {
                mappedFile = this.allocateMappedFileService.putRequestAndReturnMappedFile(nextFilePath,
                    nextNextFilePath, this.mappedFileSize);
            } else {
                try {
                  // 否则就直接创建
                    mappedFile = new MappedFile(nextFilePath, this.mappedFileSize);
                } catch (IOException e) {
                    log.error("create mappedFile exception", e);
                }
            }

            if (mappedFile != null) {
                if (this.mappedFiles.isEmpty()) {
                    mappedFile.setFirstCreateInQueue(true);
                }
                this.mappedFiles.add(mappedFile);
            }

            return mappedFile;
        }

        return mappedFileLast;
    }

首先看下直接创建的,

public MappedFile(final String fileName, final int fileSize) throws IOException {
        init(fileName, fileSize);
    }
private void init(final String fileName, final int fileSize) throws IOException {
        this.fileName = fileName;
        this.fileSize = fileSize;
        this.file = new File(fileName);
        this.fileFromOffset = Long.parseLong(this.file.getName());
        boolean ok = false;

        ensureDirOK(this.file.getParent());

        try {
          // 通过 RandomAccessFile 创建 fileChannel
            this.fileChannel = new RandomAccessFile(this.file, "rw").getChannel();
          // 做 mmap 映射
            this.mappedByteBuffer = this.fileChannel.map(MapMode.READ_WRITE, 0, fileSize);
            TOTAL_MAPPED_VIRTUAL_MEMORY.addAndGet(fileSize);
            TOTAL_MAPPED_FILES.incrementAndGet();
            ok = true;
        } catch (FileNotFoundException e) {
            log.error("create file channel " + this.fileName + " Failed. ", e);
            throw e;
        } catch (IOException e) {
            log.error("map file " + this.fileName + " Failed. ", e);
            throw e;
        } finally {
            if (!ok && this.fileChannel != null) {
                this.fileChannel.close();
            }
        }
    }

如果是提交给AllocateMappedFileService的话就用到了一些异步操作

public MappedFile putRequestAndReturnMappedFile(String nextFilePath, String nextNextFilePath, int fileSize) {
        int canSubmitRequests = 2;
        if (this.messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
            if (this.messageStore.getMessageStoreConfig().isFastFailIfNoBufferInStorePool()
                && BrokerRole.SLAVE != this.messageStore.getMessageStoreConfig().getBrokerRole()) { //if broker is slave, don't fast fail even no buffer in pool
                canSubmitRequests = this.messageStore.getTransientStorePool().remainBufferNumbs() - this.requestQueue.size();
            }
        }
				// 将请求放在 requestTable 中
        AllocateRequest nextReq = new AllocateRequest(nextFilePath, fileSize);
        boolean nextPutOK = this.requestTable.putIfAbsent(nextFilePath, nextReq) == null;
        // requestTable 使用了 concurrentHashMap,用文件名作为 key,防止并发
        if (nextPutOK) {
            // 这里判断了是否可以提交到 TransientStorePool,涉及读写分离,后面再细聊
            if (canSubmitRequests <= 0) {
                log.warn("[NOTIFYME]TransientStorePool is not enough, so create mapped file error, " +
                    "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                this.requestTable.remove(nextFilePath);
                return null;
            }
          // 塞到阻塞队列中
            boolean offerOK = this.requestQueue.offer(nextReq);
            if (!offerOK) {
                log.warn("never expected here, add a request to preallocate queue failed");
            }
            canSubmitRequests--;
        }

        // 这里的两个提交我猜测是为了多生成一个 CommitLog,
        AllocateRequest nextNextReq = new AllocateRequest(nextNextFilePath, fileSize);
        boolean nextNextPutOK = this.requestTable.putIfAbsent(nextNextFilePath, nextNextReq) == null;
        if (nextNextPutOK) {
            if (canSubmitRequests <= 0) {
                log.warn("[NOTIFYME]TransientStorePool is not enough, so skip preallocate mapped file, " +
                    "RequestQueueSize : {}, StorePoolSize: {}", this.requestQueue.size(), this.messageStore.getTransientStorePool().remainBufferNumbs());
                this.requestTable.remove(nextNextFilePath);
            } else {
                boolean offerOK = this.requestQueue.offer(nextNextReq);
                if (!offerOK) {
                    log.warn("never expected here, add a request to preallocate queue failed");
                }
            }
        }

        if (hasException) {
            log.warn(this.getServiceName() + " service has exception. so return null");
            return null;
        }

        AllocateRequest result = this.requestTable.get(nextFilePath);
        try {
          // 这里就异步等着
            if (result != null) {
                boolean waitOK = result.getCountDownLatch().await(waitTimeOut, TimeUnit.MILLISECONDS);
                if (!waitOK) {
                    log.warn("create mmap timeout " + result.getFilePath() + " " + result.getFileSize());
                    return null;
                } else {
                    this.requestTable.remove(nextFilePath);
                    return result.getMappedFile();
                }
            } else {
                log.error("find preallocate mmap failed, this never happen");
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
        }

        return null;
    }

而真正去执行文件操作的就是 AllocateMappedFileService的 run 方法

public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped() && this.mmapOperation()) {

        }
        log.info(this.getServiceName() + " service end");
    }
private boolean mmapOperation() {
        boolean isSuccess = false;
        AllocateRequest req = null;
        try {
          // 从阻塞队列里获取请求
            req = this.requestQueue.take();
            AllocateRequest expectedRequest = this.requestTable.get(req.getFilePath());
            if (null == expectedRequest) {
                log.warn("this mmap request expired, maybe cause timeout " + req.getFilePath() + " "
                    + req.getFileSize());
                return true;
            }
            if (expectedRequest != req) {
                log.warn("never expected here,  maybe cause timeout " + req.getFilePath() + " "
                    + req.getFileSize() + ", req:" + req + ", expectedRequest:" + expectedRequest);
                return true;
            }

            if (req.getMappedFile() == null) {
                long beginTime = System.currentTimeMillis();

                MappedFile mappedFile;
                if (messageStore.getMessageStoreConfig().isTransientStorePoolEnable()) {
                    try {
                      // 通过 transientStorePool 创建
                        mappedFile = ServiceLoader.load(MappedFile.class).iterator().next();
                        mappedFile.init(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    } catch (RuntimeException e) {
                        log.warn("Use default implementation.");
                      // 默认创建
                        mappedFile = new MappedFile(req.getFilePath(), req.getFileSize(), messageStore.getTransientStorePool());
                    }
                } else {
                  // 默认创建
                    mappedFile = new MappedFile(req.getFilePath(), req.getFileSize());
                }

                long eclipseTime = UtilAll.computeEclipseTimeMilliseconds(beginTime);
                if (eclipseTime > 10) {
                    int queueSize = this.requestQueue.size();
                    log.warn("create mappedFile spent time(ms) " + eclipseTime + " queue size " + queueSize
                        + " " + req.getFilePath() + " " + req.getFileSize());
                }

                // pre write mappedFile
                if (mappedFile.getFileSize() >= this.messageStore.getMessageStoreConfig()
                    .getMapedFileSizeCommitLog()
                    &&
                    this.messageStore.getMessageStoreConfig().isWarmMapedFileEnable()) {
                    mappedFile.warmMappedFile(this.messageStore.getMessageStoreConfig().getFlushDiskType(),
                        this.messageStore.getMessageStoreConfig().getFlushLeastPagesWhenWarmMapedFile());
                }

                req.setMappedFile(mappedFile);
                this.hasException = false;
                isSuccess = true;
            }
        } catch (InterruptedException e) {
            log.warn(this.getServiceName() + " interrupted, possibly by shutdown.");
            this.hasException = true;
            return false;
        } catch (IOException e) {
            log.warn(this.getServiceName() + " service has exception. ", e);
            this.hasException = true;
            if (null != req) {
                requestQueue.offer(req);
                try {
                    Thread.sleep(1);
                } catch (InterruptedException ignored) {
                }
            }
        } finally {
            if (req != null && isSuccess)
              // 通知前面等待的
                req.getCountDownLatch().countDown();
        }
        return true;
    }

这是个很大的话题了,可能会分成两部分说,第一部分就是所谓的零拷贝 ( zero-copy ),这一块其实也不新鲜,我对零拷贝的概念主要来自这篇文章,个人感觉写得非常好,在 rocketmq 中,最大的一块存储就是消息存储,也就是 CommitLog ,当然还有 ConsumeQueue 和 IndexFile,以及其他一些文件,CommitLog 的存储是以一个 1G 大小的文件作为存储单位,写完了就再建一个,那么如何提高这 1G 文件的读写效率呢,就是 mmap,传统意义的读写文件,read,write 都需要由系统调用,来回地在用户态跟内核态进行拷贝切换,

read(file, tmp_buf, len);
write(socket, tmp_buf, len);

vms95Z

如上面的图显示的,要在用户态跟内核态进行切换,数据还需要在内核缓冲跟用户缓冲之间拷贝多次,

  1. 第一步是调用 read,需要在用户态切换成内核态,DMA模块从磁盘中读取文件,并存储在内核缓冲区,相当于是第一次复制
  2. 数据从内核缓冲区被拷贝到用户缓冲区,read 调用返回,伴随着内核态又切换成用户态,完成了第二次复制
  3. 然后是write 写入,这里也会伴随着用户态跟内核态的切换,数据从用户缓冲区被复制到内核空间缓冲区,完成了第三次复制,这次有点不一样的是数据不是在内核缓冲区了,会复制到 socket buffer 中。
  4. write 系统调用返回,又切换回了用户态,然后数据由 DMA 拷贝到协议引擎。

如此就能看出其实默认的读写操作代价是非常大的,而在 rocketmq 等高性能中间件中都有使用的零拷贝技术,其中 rocketmq 使用的是 mmap

mmap

mmap基于 OS 的 mmap 的内存映射技术,通过MMU 映射文件,将文件直接映射到用户态的内存地址,使得对文件的操作不再是 write/read,而转化为直接对内存地址的操作,使随机读写文件和读写内存相似的速度。

mmap 把文件映射到用户空间里的虚拟内存,省去了从内核缓冲区复制到用户空间的过程,文件中的位置在虚拟内存中有了对应的地址,可以像操作内存一样操作这个文件,这样的文件读写文件方式少了数据从内核缓存到用户空间的拷贝,效率很高。

tmp_buf = mmap(file, len);
write(socket, tmp_buf, len);

I68mFx

第一步:mmap系统调用使得文件内容被DMA引擎复制到内核缓冲区。然后该缓冲区与用户进程共享,在内核和用户内存空间之间不进行任何拷贝。

第二步:写系统调用使得内核将数据从原来的内核缓冲区复制到与套接字相关的内核缓冲区。

第三步:第三次拷贝发生在DMA引擎将数据从内核套接字缓冲区传递给协议引擎时。

通过使用mmap而不是read,我们将内核需要拷贝的数据量减少了一半。当大量的数据被传输时,这将有很好的效果。然而,这种改进并不是没有代价的;在使用mmap+write方法时,有一些隐藏的陷阱。例如当你对一个文件进行内存映射,然后在另一个进程截断同一文件时调用写。你的写系统调用将被总线错误信号SIGBUS打断,因为你执行了一个错误的内存访问。该信号的默认行为是杀死进程并dumpcore–这对网络服务器来说不是最理想的操作。

有两种方法可以解决这个问题。

第一种方法是为SIGBUS信号安装一个信号处理程序,然后在处理程序中简单地调用返回。通过这样做,写系统调用会返回它在被打断之前所写的字节数,并将errno设置为成功。让我指出,这将是一个糟糕的解决方案,一个治标不治本的解决方案。因为SIGBUS预示着进程出了严重的问题,所以不鼓励使用这种解决方案。

第二个解决方案涉及内核的文件租赁(在Windows中称为 “机会锁”)。这是解决这个问题的正确方法。通过在文件描述符上使用租赁,你与内核在一个特定的文件上达成租约。然后你可以向内核请求一个读/写租约。当另一个进程试图截断你正在传输的文件时,内核会向你发送一个实时信号,即RT_SIGNAL_LEASE信号。它告诉你内核即将终止你对该文件的写或读租约。在你的程序访问一个无效的地址和被SIGBUS信号杀死之前,你的写调用会被打断了。写入调用的返回值是中断前写入的字节数,errno将被设置为成功。下面是一些示例代码,显示了如何从内核中获得租约。

if(fcntl(fd, F_SETSIG, RT_SIGNAL_LEASE) == -1) {
    perror("kernel lease set signal");
    return -1;
}
/* l_type can be F_RDLCK F_WRLCK */
if(fcntl(fd, F_SETLEASE, l_type)){
    perror("kernel lease set type");
    return -1;
}

rocketmq 里有一种比较特殊的用法,就是顺序消息,比如订单的生命周期里,在创建,支付,签收等状态轮转中,会发出来对应的消息,这里面就比较需要去保证他们的顺序,当然在处理的业务代码也可以做对应的处理,结合消息重投,但是如果这里消息就能保证顺序性了,那么业务代码就能更好的关注业务代码的处理。

首先有一种情况是全局的有序,比如对于一个 topic 里就得发送线程保证只有一个,内部的 queue 也只有一个,消费线程也只有一个,这样就能比较容易的保证全局顺序性了,但是这里的问题就是完全限制了性能,不是很现实,在真实场景里很多都是比如对于同一个订单,需要去保证状态的轮转是按照预期的顺序来,而不必要全局的有序性。

对于这类的有序性,需要在发送和接收方都有对应的处理,在发送消息中,需要去指定 selector,即MessageQueueSelector,能够以固定的方式是分配到对应的 MessageQueue

比如像 RocketMQ 中的示例

SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    Long id = (Long) arg;  //message queue is selected by #salesOrderID
                    long index = id % mqs.size();
                    return mqs.get((int) index);
                }
            }, orderList.get(i).getOrderId());

而在消费侧有几个点比较重要,首先我们要保证一个 MessageQueue只被一个消费者消费,消费队列存在broker端,要保证 MessageQueue 只被一个消费者消费,那么消费者在进行消息拉取消费时就必须向mq服务器申请队列锁,消费者申请队列锁的代码存在于RebalanceService消息队列负载的实现代码中。

List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
        for (MessageQueue mq : mqSet) {
            if (!this.processQueueTable.containsKey(mq)) {
              // 判断是否顺序,如果是顺序消费的,则需要加锁
                if (isOrder && !this.lock(mq)) {
                    log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
                    continue;
                }

                this.removeDirtyOffset(mq);
                ProcessQueue pq = new ProcessQueue();
                long nextOffset = this.computePullFromWhere(mq);
                if (nextOffset >= 0) {
                    ProcessQueue pre = this.processQueueTable.putIfAbsent(mq, pq);
                    if (pre != null) {
                        log.info("doRebalance, {}, mq already exists, {}", consumerGroup, mq);
                    } else {
                        log.info("doRebalance, {}, add a new mq, {}", consumerGroup, mq);
                        PullRequest pullRequest = new PullRequest();
                        pullRequest.setConsumerGroup(consumerGroup);
                        pullRequest.setNextOffset(nextOffset);
                        pullRequest.setMessageQueue(mq);
                        pullRequest.setProcessQueue(pq);
                        pullRequestList.add(pullRequest);
                        changed = true;
                    }
                } else {
                    log.warn("doRebalance, {}, add new mq failed, {}", consumerGroup, mq);
                }
            }
        }

在申请到锁之后会创建 pullRequest 进行消息拉取,消息拉取部分的代码实现在PullMessageService中,

@Override
    public void run() {
        log.info(this.getServiceName() + " service started");

        while (!this.isStopped()) {
            try {
                PullRequest pullRequest = this.pullRequestQueue.take();
                this.pullMessage(pullRequest);
            } catch (InterruptedException ignored) {
            } catch (Exception e) {
                log.error("Pull Message Service Run Method exception", e);
            }
        }

        log.info(this.getServiceName() + " service end");
    }

消息拉取完后,需要提交到ConsumeMessageService中进行消费,顺序消费的实现为ConsumeMessageOrderlyService,提交消息进行消费的方法为ConsumeMessageOrderlyService#submitConsumeRequest,具体实现如下:

@Override
public void submitConsumeRequest(
    final List<MessageExt> msgs,
    final ProcessQueue processQueue,
    final MessageQueue messageQueue,
    final boolean dispathToConsume) {
    if (dispathToConsume) {
        ConsumeRequest consumeRequest = new ConsumeRequest(processQueue, messageQueue);
        this.consumeExecutor.submit(consumeRequest);
    }
}

构建了一个ConsumeRequest对象,它是个实现了 runnable 接口的类,并提交给了线程池来并行消费,看下顺序消费的ConsumeRequest的run方法实现:

@Override
        public void run() {
            if (this.processQueue.isDropped()) {
                log.warn("run, the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                return;
            }
						// 获得 Consumer 消息队列锁,即单个线程独占
            final Object objLock = messageQueueLock.fetchLockObject(this.messageQueue);
            synchronized (objLock) {
              // (广播模式) 或者 (集群模式 && Broker消息队列锁有效)
                if (MessageModel.BROADCASTING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                    || (this.processQueue.isLocked() && !this.processQueue.isLockExpired())) {
                    final long beginTime = System.currentTimeMillis();
                  // 循环
                    for (boolean continueConsume = true; continueConsume; ) {
                        if (this.processQueue.isDropped()) {
                            log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                            break;
                        }

                      // 消息队列分布式锁未锁定,提交延迟获得锁并消费请求
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && !this.processQueue.isLocked()) {
                            log.warn("the message queue not locked, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }

                      // 消息队列分布式锁已经过期,提交延迟获得锁并消费请求
                        if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())
                            && this.processQueue.isLockExpired()) {
                            log.warn("the message queue lock expired, so consume later, {}", this.messageQueue);
                            ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 10);
                            break;
                        }
												// 当前周期消费时间超过连续时长,默认:60s,提交延迟消费请求。默认情况下,每消费1分钟休息10ms。
                        long interval = System.currentTimeMillis() - beginTime;
                        if (interval > MAX_TIME_CONSUME_CONTINUOUSLY) {
                            ConsumeMessageOrderlyService.this.submitConsumeRequestLater(processQueue, messageQueue, 10);
                            break;
                        }
												// 获取消费消息。此处和并发消息请求不同,并发消息请求已经带了消费哪些消息。
                        final int consumeBatchSize =
                            ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumeMessageBatchMaxSize();

                        List<MessageExt> msgs = this.processQueue.takeMessags(consumeBatchSize);
                        defaultMQPushConsumerImpl.resetRetryAndNamespace(msgs, defaultMQPushConsumer.getConsumerGroup());
                        if (!msgs.isEmpty()) {
                            final ConsumeOrderlyContext context = new ConsumeOrderlyContext(this.messageQueue);

                            ConsumeOrderlyStatus status = null;

                            ConsumeMessageContext consumeMessageContext = null;
                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext = new ConsumeMessageContext();
                                consumeMessageContext
                                    .setConsumerGroup(ConsumeMessageOrderlyService.this.defaultMQPushConsumer.getConsumerGroup());
                                consumeMessageContext.setNamespace(defaultMQPushConsumer.getNamespace());
                                consumeMessageContext.setMq(messageQueue);
                                consumeMessageContext.setMsgList(msgs);
                                consumeMessageContext.setSuccess(false);
                                // init the consume context type
                                consumeMessageContext.setProps(new HashMap<String, String>());
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookBefore(consumeMessageContext);
                            }
														// 执行消费
                            long beginTimestamp = System.currentTimeMillis();
                            ConsumeReturnType returnType = ConsumeReturnType.SUCCESS;
                            boolean hasException = false;
                            try {
                                this.processQueue.getLockConsume().lock(); // 锁定处理队列
                                if (this.processQueue.isDropped()) {
                                    log.warn("consumeMessage, the message queue not be able to consume, because it's dropped. {}",
                                        this.messageQueue);
                                    break;
                                }

                                status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context);
                            } catch (Throwable e) {
                                log.warn("consumeMessage exception: {} Group: {} Msgs: {} MQ: {}",
                                    RemotingHelper.exceptionSimpleDesc(e),
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                                hasException = true;
                            } finally {
                                this.processQueue.getLockConsume().unlock();  // 解锁
                            }

                            if (null == status
                                || ConsumeOrderlyStatus.ROLLBACK == status
                                || ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                log.warn("consumeMessage Orderly return not OK, Group: {} Msgs: {} MQ: {}",
                                    ConsumeMessageOrderlyService.this.consumerGroup,
                                    msgs,
                                    messageQueue);
                            }

                            long consumeRT = System.currentTimeMillis() - beginTimestamp;
                            if (null == status) {
                                if (hasException) {
                                    returnType = ConsumeReturnType.EXCEPTION;
                                } else {
                                    returnType = ConsumeReturnType.RETURNNULL;
                                }
                            } else if (consumeRT >= defaultMQPushConsumer.getConsumeTimeout() * 60 * 1000) {
                                returnType = ConsumeReturnType.TIME_OUT;
                            } else if (ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT == status) {
                                returnType = ConsumeReturnType.FAILED;
                            } else if (ConsumeOrderlyStatus.SUCCESS == status) {
                                returnType = ConsumeReturnType.SUCCESS;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.getProps().put(MixAll.CONSUME_CONTEXT_TYPE, returnType.name());
                            }

                            if (null == status) {
                                status = ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                            }

                            if (ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.hasHook()) {
                                consumeMessageContext.setStatus(status.toString());
                                consumeMessageContext
                                    .setSuccess(ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status);
                                ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.executeHookAfter(consumeMessageContext);
                            }

                            ConsumeMessageOrderlyService.this.getConsumerStatsManager()
                                .incConsumeRT(ConsumeMessageOrderlyService.this.consumerGroup, messageQueue.getTopic(), consumeRT);

                            continueConsume = ConsumeMessageOrderlyService.this.processConsumeResult(msgs, status, context, this);
                        } else {
                            continueConsume = false;
                        }
                    }
                } else {
                    if (this.processQueue.isDropped()) {
                        log.warn("the message queue not be able to consume, because it's dropped. {}", this.messageQueue);
                        return;
                    }

                    ConsumeMessageOrderlyService.this.tryLockLaterAndReconsume(this.messageQueue, this.processQueue, 100);
                }
            }
        }

获取到锁对象后,使用synchronized尝试申请线程级独占锁。

如果加锁成功,同一时刻只有一个线程进行消息消费。

如果加锁失败,会延迟100ms重新尝试向broker端申请锁定messageQueue,锁定成功后重新提交消费请求

创建消息拉取任务时,消息客户端向broker端申请锁定MessageQueue,使得一个MessageQueue同一个时刻只能被一个消费客户端消费。

消息消费时,多线程针对同一个消息队列的消费先尝试使用synchronized申请独占锁,加锁成功才能进行消费,使得一个MessageQueue同一个时刻只能被一个消费客户端中一个线程消费。
这里其实还有很重要的一点是对processQueue的加锁,这里其实是保证了在 rebalance的过程中如果 processQueue 被分配给了另一个 consumer,但是当前已经被我这个 consumer 再消费,但是没提交,就有可能出现被两个消费者消费,所以得进行加锁保证不受 rebalance 影响。

前面主要还是说了乒乓球的,因为整体还是乒乓球的比赛赛程比较长,比较激烈,扣人心弦,记得那会在公司没法看视频直播,就偶尔看看奥运会官网的比分,还几场马龙樊振东,陈梦被赢了一局就吓尿了,已经被混双那场留下了阴影,其实后面去看看16 年的比赛什么的,中国队虽然拿了这么多冠军,但是自改成 11 分制以来,其实都没办法那么完全彻底地碾压,而且像张继科,樊振东,陈梦都多少有些慢热,现在看来是马龙比较全面,不过看过了马龙,刘国梁,许昕等的一些过往经历,都是起起伏伏,即使是张怡宁这样的大魔王,也经历过逢王楠不赢的阶段,心态无法调整好。

其实最开始是举重项目,侯志慧是女子 49 公斤级的冠军,这场比赛是全场都看,其实看中国队的举重比赛跟跳水有点像,每一轮都需要到最后才能等到中国队,跳水其实每轮都有,举重会按照自己报的试举重量进行排名,重量大的会在后面举,抓举和挺举各三次试举机会,有时候会看着比较焦虑,一直等不来,怕一上来就没试举成功,而且中国队一般试举重量就是很大的,容易一次试举不成功就马上下一次,连着举其实压力会非常大,说实话真的是外行看热闹,每次都是多懂一点点,这次由于实在是比较无聊,所以看的会比较专心点,对于对应的规则知识点也会多了解一点,同时对于举重,没想到我们国家的这些运动员有这么强,最后八块金牌拿了七块,有一块拿到银牌也是有点因为教练的策略问题,这里其实也稍微知道一点,因为报上去的试举重量是谁小谁先举,并且我们国家都是实力非常强的,所以都会报大一些,并且如果这个项目有实力相近的选手,会比竞对多报一公斤,这样子如果前面竞争对手没举成功,我们把握就很大了,最坏的情况即使对手试举成功了,我们还有机会搏一把,比如谌利军这样的,只是说说感想,举重运动员真的是个比较单纯的群体,而且训练是非常痛苦枯燥的,非常容易受伤,像挺举就有点会压迫呼吸通道,看到好几个都是脸憋得通红,甚至直接因为压迫气道而没法完成后面的挺举,像之前 16 年的举重比赛,有个运动员没成功夺冠就非常愧疚地哭着说对不起祖国,没有获得冠军,这是怎么样的一种歉疚,怎么样的一种纯粹的感情呢,相对应地来说,我又要举男足,男篮的例子了,很多人在那嘲笑我这样对男足男篮愤愤不平的人,说可能我这样的人都没交个税(从缴纳个税的数量比例来算有可能),只是这里有两个打脸的事情,我足额缴纳个税,接近 20%的薪资都缴了个税,并且我买的所有东西都缴了增值税,如果让我这样缴纳了个税,缴纳了增值税的有个人的投票权,我一定会投票不让男足男篮使用我缴纳我的税金,用我们的缴纳的税,打出这么烂的表现,想乒乓球混双,拿个亚军都会被喷,那可是世界第二了,而且是就输了那么一场,足球篮球呢,我觉得是一方面成绩差,因为比赛真的有状态跟心态的影响,偶尔有一场失误非常正常,NBA 被黑八的有这么多强队,但是如果像男足男篮,成绩是越来越差,用范志毅的话来说就是脸都不要了,还有就是精气神,要在比赛中打出胜负欲,保持这种争胜心,才有机会再进步,前火箭队主教练鲁迪·汤姆贾诺维奇的话,“永远不要低估冠军的决心”,即使我现在打不过你,我会在下一次,下下次打败你,竞技体育永远要有这种精神,可以接受一时的失败,但是要保持永远争胜的心。

第一块金牌是杨倩拿下的,中国队拿奥运会首金也是有政治任务的,而恰恰杨倩这个金牌也有点碰巧是对手最后一枪失误了,当然竞技体育,特别是射击,真的是容不得一点点失误,像前面几届的美国神通埃蒙斯,失之毫厘差之千里,但是这个具体评价就比较少,唯一一点让我比较出戏的就是杨倩真的非常像王刚的徒弟漆二娃,哈哈,微博上也有挺多人觉得像,射击还是个比较可以接受年纪稍大的运动员,需要经验和稳定性,相对来说爆发力体力稍好一点,像庞伟这样的,混合团体10米气手枪金牌,36 岁可能其他项目已经是年龄很大了,不过前面说的举重的吕小军军神也是年纪蛮大了,但是非常强,而且在油管上简直就是个神,相对来说射击是关注比较少,杨倩的也只是看了后面拿到冠军这个结果,有些因为时间或者电视上没放,但是成绩还是不错的,没多少喷点。

第二篇先到这,纯主观,轻喷。

redis 在使用 rdb 策略进行备份时,rdb 的意思是会在开启备份的时候将开启时间点的内存数据进行备份,并且可以设置时间,这样子就是这个策略其实还是不完全可靠的,如果是在这个间隔中宕机了,或者间隔过长,不过这个不在这次的要说的内容中,如果自己去写这个 rdb 的策略可能就有点类似于 mvcc 的 redolog,需要知道这个时间点之前的数据是怎么样的,防止后面更改的干扰,但是这样一方面需要有比较复杂的 mvcc 实现,另一方面是很占用存储空间,所以 redis 在这里面使用了 COW (Copy On Write) 技术,这个技术呢以前听过,也大致了解是怎么个意思,这次稍微具体地来看下,其实 redis 的 copy-on-write 就是来自于 linux 的 cow

Linux中的CopyOnWrite

fork()之后,kernel把父进程中所有的内存页的权限都设为read-only,然后子进程的地址空间指向父进程。当父子进程都只读内存时,相安无事。当其中某个进程写内存时,CPU硬件检测到内存页是read-only的,于是触发页异常中断(page-fault),陷入kernel的一个中断例程。中断例程中,kernel就会把触发的异常的页复制一份,于是父子进程各自持有独立的一份。这个操作其实可以类比为写屏障,正常的读取是没问题的,当有写入时就会分裂。

CopyOnWrite的好处:

1、减少分配和复制资源时带来的瞬时延迟;
2、减少不必要的资源分配;
CopyOnWrite的缺点:
1、如果父子进程都需要进行大量的写操作,会产生大量的分页错误(页异常中断page-fault);

Redis中的CopyOnWrite

Redis在持久化时,如果是采用BGSAVE命令或者BGREWRITEAOF的方式,那Redis会fork出一个子进程来读取数据,从而写到磁盘中。
总体来看,Redis还是读操作比较多。如果子进程存在期间,发生了大量的写操作,那可能就会出现很多的分页错误(页异常中断page-fault),这样就得耗费不少性能在复制上。
而在rehash阶段上,写操作是无法避免的。所以Redis在fork出子进程之后,将负载因子阈值提高,尽量减少写操作,避免不必要的内存写入操作,最大限度地节约内存。这里其实更巧妙了,在细节上去优化会产生大量页异常中断的情况。