[metaq]Broker

Broker是metaq的核心组件，负责消息的物理存储，分区指定等。例子配置文件

[system]
#broker编号，集群唯一
brokerId=0
#这个broker指定的分区数
numPartitions=2
#nio port
serverPort=8123
#异步刷盘策略，为0表示同步刷盘
unflushThreshold=0
#同上，刷盘间隔
unflushInterval=10000
#单个消息文件的最大size
maxSegmentSize=1073741824
#单个请求最大传输size
maxTransferSize=1048576
#数据清理
deletePolicy=delete,168
#quartz的delete任务表达式
deleteWhen=0 0 6,18 * * ?
flushTxLogAtCommit=1
stat=true

;; Update consumers offsets to current max offsets when consumers offsets are out of range of current broker's messages.
;; It must be false in production.But recommend to be true in development or test.
updateConsumerOffsets=true

[zookeeper]
zk.zkConnect=localhost:2181
zk.zkSessionTimeoutMs=30000
zk.zkConnectionTimeoutMs=30000
zk.zkSyncTimeMs=5000

;; Topics section
#topic名称
[topic=test]

[topic=meta-test]

 broker核心类图如下

 

Broker处理如下

1.broker启动

MetaMorphosisBroker创建，初始化zk连接

public MetaMorphosisBroker(final MetaConfig metaConfig) {
        super();
        this.metaConfig = metaConfig;
	//NIO server
        this.remotingServer = newRemotingServer(metaConfig);
        this.executorsManager = new ExecutorsManager(metaConfig);
	//全局ID生成器
        this.idWorker = new IdWorker(metaConfig.getBrokerId());
	//文件管理
        this.storeManager = new MessageStoreManager(metaConfig, this.newDeletePolicy(metaConfig));
	//监控
        this.statsManager = new StatsManager(this.metaConfig, this.storeManager, this.remotingServer);
	//zookeeper客户端
        this.brokerZooKeeper = new BrokerZooKeeper(metaConfig);
        final BrokerCommandProcessor next =
                new BrokerCommandProcessor(this.storeManager, this.executorsManager, this.statsManager,
                    this.remotingServer, metaConfig, this.idWorker, this.brokerZooKeeper);
        JournalTransactionStore transactionStore = null;
        try {
            transactionStore = new JournalTransactionStore(metaConfig.getDataLogPath(), this.storeManager, metaConfig);
        }
        catch (final Exception e) {
            throw new MetamorphosisServerStartupException("Initializing transaction store failed", e);
        }
	//带事务的处理器
        this.brokerProcessor =
                new TransactionalCommandProcessor(metaConfig, this.storeManager, this.idWorker, next, transactionStore,
                    this.statsManager);
        this.shutdownHook = new ShutdownHook();
        Runtime.getRuntime().addShutdownHook(this.shutdownHook);
        MetaMBeanServer.registMBean(this, null);
    }

 broker启动

public synchronized void start() {
        if (!this.shutdown) {
            return;
        }
        this.shutdown = false;
	//加载已有数据并校验老数据
        this.storeManager.init();
        this.executorsManager.init();
        this.statsManager.init();
	//不同command对应不同处理器
        this.registerProcessors();
        try {
		//NIO server启动
            this.remotingServer.start();
        }
        catch (final NotifyRemotingException e) {
            throw new MetamorphosisServerStartupException("start remoting server failed", e);
        }
        try {
		//在/brokers/ids下创建临时节点，名称为节点Id
		//如果为master节点，则创建/brokers/ids/0/master节点
            this.brokerZooKeeper.registerBrokerInZk();
		//如果为master节点，则创建/brokers/ids/master_config_checksum节点
            this.brokerZooKeeper.registerMasterConfigFileChecksumInZk();
            this.addTopicsChangeListener();
		//在/brokers/topics-sub和/brokers/pub创建对应topic临时节点-topics
            this.registerTopicsInZk();
            this.registerZkSuccess = true;
        }
        catch (final Exception e) {
            this.registerZkSuccess = false;
            throw new MetamorphosisServerStartupException("Register broker to zk failed", e);
        }
        log.info("Starting metamorphosis server...");
        this.brokerProcessor.init();
        log.info("Start metamorphosis server successfully");
    }

 注册的处理器

private void registerProcessors() {
        this.remotingServer.registerProcessor(GetCommand.class, new GetProcessor(this.brokerProcessor,
            this.executorsManager.getGetExecutor()));
        this.remotingServer.registerProcessor(PutCommand.class, new PutProcessor(this.brokerProcessor,
            this.executorsManager.getUnOrderedPutExecutor()));
        this.remotingServer.registerProcessor(OffsetCommand.class, new OffsetProcessor(this.brokerProcessor,
            this.executorsManager.getGetExecutor()));
        this.remotingServer
        .registerProcessor(HeartBeatRequestCommand.class, new VersionProcessor(this.brokerProcessor));
        this.remotingServer.registerProcessor(QuitCommand.class, new QuitProcessor(this.brokerProcessor));
        this.remotingServer.registerProcessor(StatsCommand.class, new StatsProcessor(this.brokerProcessor));
        this.remotingServer.registerProcessor(TransactionCommand.class, new TransactionProcessor(this.brokerProcessor,
            this.executorsManager.getUnOrderedPutExecutor()));
    }

 2.无事务put请求

根据注册的处理器，put请求由PutProcessor处理

 public void handleRequest(final PutCommand request, final Connection conn) {
        final TransactionId xid = request.getTransactionId();
        final SessionContext context = SessionContextHolder.getOrCreateSessionContext(conn, xid);
        try {
            this.processor.processPutCommand(request, context, new PutCallback() {
                @Override
                public void putComplete(final ResponseCommand resp) {
                    RemotingUtils.response(context.getConnection(), resp);
                }
            });
            // RemotingUtils.response(context.getConnection(),
            // PutProcessor.this.processor.processPutCommand(request, context));
        }
        catch (final Exception e) {
            RemotingUtils.response(context.getConnection(), new BooleanCommand(HttpStatus.InternalServerError,
                e.getMessage(), request.getOpaque()));
        }
    }

 无事务请求由BrokerCommandProcessor处理

 public void processPutCommand(final PutCommand request, final SessionContext sessionContext, final PutCallback cb) {
        final String partitionString = this.metaConfig.getBrokerId() + "-" + request.getPartition();
      	.....
		//partition信息
            final int partition = this.getPartition(request);
		//partition对应的store
            final MessageStore store = this.storeManager.getOrCreateMessageStore(request.getTopic(), partition);
            // 如果是动态添加的topic，需要注册到zk
            this.brokerZooKeeper.registerTopicInZk(request.getTopic(), false);
            // 设置唯一id
            final long messageId = this.idWorker.nextId();
		//写数据
            store.append(messageId, request,
                new StoreAppendCallback(partition, partitionString, request, messageId, cb));
        }
        catch (final Exception e) {
            this.statsManager.statsPutFailed(request.getTopic(), partitionString, 1);
            log.error("Put message failed", e);
            if (cb != null) {
                cb.putComplete(new BooleanCommand(HttpStatus.InternalServerError, e.getMessage(), request.getOpaque()));
            }
        }
    }

 具体写数据过程

private void appendBuffer(final ByteBuffer buffer, final AppendCallback cb) {
        if (this.closed) {
            throw new IllegalStateException("Closed MessageStore.");
        }
        if (this.useGroupCommit() && buffer.remaining() < this.maxTransferSize) {
            this.bufferQueue.offer(new WriteRequest(buffer, cb));
        }
        else {
            Location location = null;
            final int remainning = buffer.remaining();
		//单线程append
            this.writeLock.lock();
            try {
		//当前最新的消息文件
                final Segment cur = this.segments.last();
		//append，返回写入数据量
                final long offset = cur.start + cur.fileMessageSet.append(buffer);
		//根据刷盘策略，判断需要刷盘
                this.mayBeFlush(1);
		//超过一定大小，生成新的消息文件
                this.mayBeRoll();
                location = new Location(offset, remainning);
            }
            catch (final IOException e) {
                log.error("Append file failed", e);
                location = Location.InvalidLocaltion;
            }
            finally {
                this.writeLock.unlock();
                if (cb != null) {
                    cb.appendComplete(location);
                }
            }
        }
    }

 append

public long append(final ByteBuffer buf) throws IOException {
        if (!this.mutable) {
            throw new UnsupportedOperationException("Immutable message set");
        }
        final long offset = this.sizeInBytes.get();
        int sizeInBytes = 0;
	//循环写入buffer，直到完成
        while (buf.hasRemaining()) {
            sizeInBytes += this.channel.write(buf);
        }
        this.sizeInBytes.addAndGet(sizeInBytes);
        this.messageCount.incrementAndGet();
        return offset;
    }

 flush

  public void flush() throws IOException {
	//写入磁盘
        this.channel.force(true);
        this.highWaterMark.set(this.sizeInBytes.get());
    }

 3.get请求

根据注册的处理器，get请求由GetProcessor处理，get请求无事务，BrokerCommandProcessor处理

public ResponseCommand processGetCommand(final GetCommand request, final SessionContext ctx, final boolean zeroCopy) {
        final String group = request.getGroup();
        final String topic = request.getTopic();
        this.statsManager.statsGet(topic, group, 1);

        // 如果分区被关闭,禁止读数据 --wuhua
        if (this.metaConfig.isClosedPartition(topic, request.getPartition())) {
            log.warn("can not get message for topic=" + topic + " from partition " + request.getPartition()
                + ",it closed,");
            return new BooleanCommand(HttpStatus.Forbidden, "Partition["
                    + this.metaConfig.getBrokerId() + "-" + request.getPartition() + "] has been closed", request.getOpaque());
        }
	//根据partition拿store
        final MessageStore store = this.storeManager.getMessageStore(topic, request.getPartition());
        if (store == null) {
            this.statsManager.statsGetMiss(topic, group, 1);
            return new BooleanCommand(HttpStatus.NotFound, "The topic `" + topic
                + "` in partition `" + request.getPartition() + "` is not exists", request.getOpaque());
        }
        if (request.getMaxSize() <= 0) {
            return new BooleanCommand(HttpStatus.BadRequest, "Bad request,invalid max size:"
                    + request.getMaxSize(), request.getOpaque());
        }
        try {
		//根据offset和transferSize拿messageSet视图
            final MessageSet set =
                    store.slice(request.getOffset(),
                        Math.min(this.metaConfig.getMaxTransferSize(), request.getMaxSize()));
            if (set != null) {
		//zeroCopy直接从OS内核写入socket缓存，不经过用户态
                if (zeroCopy) {
                    set.write(request, ctx);
                    return null;
                }
                else {
                    // refer to the code of line 440 in MessageStore
                    // create two copies of byte array including the byteBuffer
                    // and new bytes
                    // this may not a good use case of Buffer
                    final ByteBuffer byteBuffer =
                            ByteBuffer.allocate(Math.min(this.metaConfig.getMaxTransferSize(), request.getMaxSize()));
                    set.read(byteBuffer);
                    byteBuffer.flip();
                    final byte[] bytes = new byte[byteBuffer.remaining()];
                    byteBuffer.get(bytes);
                    return new DataCommand(bytes, request.getOpaque());
                }
            }
		//没取到数据。。。
            else {
                this.statsManager.statsGetMiss(topic, group, 1);
                this.statsManager.statsGetFailed(topic, group, 1);

                // 当请求的偏移量大于实际最大值时,返回给客户端实际最大的偏移量.
                final long maxOffset = store.getMaxOffset();
                final long requestOffset = request.getOffset();
                if (requestOffset > maxOffset
                        && (this.metaConfig.isUpdateConsumerOffsets() || requestOffset == Long.MAX_VALUE)) {
                    log.info("offset[" + requestOffset + "] is exceeded,tell the client real max offset: " + maxOffset
                        + ",topic=" + topic + ",group=" + group);
                    this.statsManager.statsOffset(topic, group, 1);
                    return new BooleanCommand(HttpStatus.Moved, String.valueOf(maxOffset), request.getOpaque());
                }
                else {
                    return new BooleanCommand(HttpStatus.NotFound, "Could not find message at position " + requestOffset,
                        request.getOpaque());
                }
            }
        }
        catch (final ArrayIndexOutOfBoundsException e) {
            log.error("Could not get message from position " + request.getOffset() + ",it is out of bounds,topic="
                    + topic);
            // 告知最近可用的offset
            this.statsManager.statsGetMiss(topic, group, 1);
            this.statsManager.statsGetFailed(topic, group, 1);
            final long validOffset = store.getNearestOffset(request.getOffset());
            this.statsManager.statsOffset(topic, group, 1);
            return new BooleanCommand(HttpStatus.Moved, String.valueOf(validOffset), request.getOpaque());

        }
        catch (final Throwable e) {
            log.error("Could not get message from position " + request.getOffset(), e);
            this.statsManager.statsGetFailed(topic, group, 1);
            return new BooleanCommand(HttpStatus.InternalServerError, e.getMessage(), request.getOpaque());
        }

    }

 slice获取message视图

    /**
     * 根据offset和maxSize返回所在MessageSet, 当offset超过最大offset的时候返回null，
     * 当offset小于最小offset的时候抛出ArrayIndexOutOfBounds异常
     * 
     * @param offset
     * 
     * @param maxSize
     * @return
     * @throws IOException
     */
    public MessageSet slice(final long offset, final int maxSize) throws IOException {
	//二分查找命中segment
        final Segment segment = this.findSegment(this.segments.view(), offset);
        if (segment == null) {
            return null;
        }
        else {
		//返回message视图
            return segment.fileMessageSet.slice(offset - segment.start, offset - segment.start + maxSize);
        }
    }