您好,登录后才能下订单哦!
这篇文章主要介绍“Apache Pulsar二进制协议怎么实现”,在日常操作中,相信很多人在Apache Pulsar二进制协议怎么实现问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Apache Pulsar二进制协议怎么实现”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
pulsar 使用protocolBuf 作为二进制协议编写的工具
message BaseCommand { enum Type { CONNECT = 2; CONNECTED = 3; // consumer 注册 SUBSCRIBE = 4; // producer 注册 PRODUCER = 5; // 向topic写入消息 SEND = 6; // 写入的response SEND_RECEIPT= 7; // 写入异常的response SEND_ERROR = 8; // 发message 给consumer MESSAGE = 9; // 确认某个消息是否成功消费 ACK = 10; // consumer 请求消息 FLOW = 11; UNSUBSCRIBE = 12; // 通用的一个成功的response SUCCESS = 13; // 通用的一个异常的response ERROR = 14; CLOSE_PRODUCER = 15; CLOSE_CONSUMER = 16; // Producer 的 response PRODUCER_SUCCESS = 17; // 网络层keepAlive 用的 PING = 18; PONG = 19; // REDELIVER_UNACKNOWLEDGED_MESSAGES = 20; PARTITIONED_METADATA = 21; PARTITIONED_METADATA_RESPONSE = 22; LOOKUP = 23; LOOKUP_RESPONSE = 24; CONSUMER_STATS = 25; CONSUMER_STATS_RESPONSE = 26; // REACHED_END_OF_TOPIC = 27; SEEK = 28; GET_LAST_MESSAGE_ID = 29; GET_LAST_MESSAGE_ID_RESPONSE = 30; // ACTIVE_CONSUMER_CHANGE = 31; GET_TOPICS_OF_NAMESPACE = 32; GET_TOPICS_OF_NAMESPACE_RESPONSE = 33; GET_SCHEMA = 34; GET_SCHEMA_RESPONSE = 35; AUTH_CHALLENGE = 36; AUTH_RESPONSE = 37; ACK_RESPONSE = 38; GET_OR_CREATE_SCHEMA = 39; GET_OR_CREATE_SCHEMA_RESPONSE = 40; // transaction related // 事务相关的比较容易理解,下面先忽略了 50 - 61 } // ..... }
这里是客户端与server连接的channel一连上就会发送一个CONNECT 请求
这里会有一些鉴权和协议版本上报的信息。
沟通客户端版本之后,服务端就知道客户端支持哪些特性,会做一些兼容处理
相当于kafka 里面的ApiVersionRequest
// org.apache.pulsar.client.impl.ClientCnx public void channelActive(ChannelHandlerContext ctx) throws Exception { super.channelActive(ctx); this.timeoutTask = this.eventLoopGroup.scheduleAtFixedRate(() -> checkRequestTimeout(), operationTimeoutMs, operationTimeoutMs, TimeUnit.MILLISECONDS); if (proxyToTargetBrokerAddress == null) { if (log.isDebugEnabled()) { log.debug("{} Connected to broker", ctx.channel()); } } else { log.info("{} Connected through proxy to target broker at {}", ctx.channel(), proxyToTargetBrokerAddress); } // Send CONNECT command ctx.writeAndFlush(newConnectCommand()) .addListener(future -> { if (future.isSuccess()) { if (log.isDebugEnabled()) { log.debug("Complete: {}", future.isSuccess()); } state = State.SentConnectFrame; } else { log.warn("Error during handshake", future.cause()); ctx.close(); } }); }
这里实际上是CommandConnect 的response ,但是换了名字
(很容易对不上号)
// org.apache.pulsar.broker.service.ServerCnx protected void handleConnect(CommandConnect connect) { checkArgument(state == State.Start); if (log.isDebugEnabled()) { log.debug("Received CONNECT from {}, auth enabled: {}:" + " has original principal = {}, original principal = {}", remoteAddress, service.isAuthenticationEnabled(), connect.hasOriginalPrincipal(), connect.getOriginalPrincipal()); } String clientVersion = connect.getClientVersion(); int clientProtocolVersion = connect.getProtocolVersion(); features = new FeatureFlags(); if (connect.hasFeatureFlags()) { features.copyFrom(connect.getFeatureFlags()); } if (!service.isAuthenticationEnabled()) { completeConnect(clientProtocolVersion, clientVersion); return; } // ...... }
这个RPC是consumer用来在服务端注册的。
具体调用的位置是,在ConsumerImpl
构造函数的最后一行会请求服务端和客户端进行连接,如果拿到了一个Connection,会调用这个连接成功的回调connectionOpened
如果是consumer的话就会发送这个请求,来注册consumer相关的信息。
如果和上面的CommandConnect
请求联动起来,这个请求是在CommandConnect
之后发送的。
// org.apache.pulsar.client.impl.ConsumerImpl @Override public void connectionOpened(final ClientCnx cnx) { // ... 上面做了一大堆的准备参数先忽略 // 构建一个subscription ByteBuf request = Commands.newSubscribe(topic, subscription, consumerId, requestId, getSubType(), priorityLevel, consumerName, isDurable, startMessageIdData, metadata, readCompacted, conf.isReplicateSubscriptionState(), InitialPosition.valueOf(subscriptionInitialPosition.getValue()), startMessageRollbackDuration, schemaInfo, createTopicIfDoesNotExist, conf.getKeySharedPolicy()); }
proto定义说明(见注释)
message CommandSubscribe { // 这里对应subscription的4种类型 enum SubType { Exclusive = 0; Shared = 1; Failover = 2; Key_Shared = 3; } // topic 名字 required string topic = 1; // subscription 名字 required string subscription = 2; // subscription 类型 required SubType subType = 3; // 这个是用来标记这个网络连接上的consumer标识 required uint64 consumer_id = 4; // 网络层的请求标识 required uint64 request_id = 5; // consumer 名字 optional string consumer_name = 6; // consumer 的优先级,优先级高的consumer 容易先收到请求 optional int32 priority_level = 7; // 这个subsciption是否是持久化的 // Signal wether the subscription should be backed by a // durable cursor or not optional bool durable = 8 [default = true]; // If specified, the subscription will position the cursor // markd-delete position on the particular message id and // will send messages from that point optional MessageIdData start_message_id = 9; // 加了一些consumer 的自定义tag Map<String,String> /// Add optional metadata key=value to this consumer repeated KeyValue metadata = 10; optional bool read_compacted = 11; optional Schema schema = 12; // 初始化位置从哪里开始,最新还是最旧 enum InitialPosition { Latest = 0; Earliest = 1; } // Signal whether the subscription will initialize on latest // or not -- earliest optional InitialPosition initialPosition = 13 [default = Latest]; // geo-replication 相关,先忽略 // Mark the subscription as "replicated". Pulsar will make sure // to periodically sync the state of replicated subscriptions // across different clusters (when using geo-replication). optional bool replicate_subscription_state = 14; // If true, the subscribe operation will cause a topic to be // created if it does not exist already (and if topic auto-creation // is allowed by broker. // If false, the subscribe operation will fail if the topic // does not exist. optional bool force_topic_creation = 15 [default = true]; // 这个是按照时间重置消费进度的时候 // If specified, the subscription will reset cursor's position back // to specified seconds and will send messages from that point optional uint64 start_message_rollback_duration_sec = 16 [default = 0]; // key_Share 模式使用的,暂时不看 optional KeySharedMeta keySharedMeta = 17; }
这个RPC 和 consumer相对应的,是producer在服务端注册用的,调用位置也是相同的org.apache.pulsar.client.impl.ProducerImpl.connectionOpened
里面。
/// Create a new Producer on a topic, assigning the given producer_id, /// all messages sent with this producer_id will be persisted on the topic message CommandProducer { // topic required string topic = 1; required uint64 producer_id = 2; // 网络层的请求编号 required uint64 request_id = 3; /// If a producer name is specified, the name will be used, /// otherwise the broker will generate a unique name optional string producer_name = 4; // 是否是加密的写入 optional bool encrypted = 5 [default = false]; // 元数据 Map<String,String> /// Add optional metadata key=value to this producer repeated KeyValue metadata = 6; optional Schema schema = 7; // 这里应该叫producer_epoch // If producer reconnect to broker, the epoch of this producer will +1 optional uint64 epoch = 8 [default = 0]; // Indicate the name of the producer is generated or user provided // Use default true here is in order to be forward compatible with the client optional bool user_provided_producer_name = 9 [default = true]; // 这里是写入的3种方式 // Require that this producers will be the only producer allowed on the topic optional ProducerAccessMode producer_access_mode = 10 [default = Shared]; // Topic epoch is used to fence off producers that reconnects after a new // exclusive producer has already taken over. This id is assigned by the // broker on the CommandProducerSuccess. The first time, the client will // leave it empty and then it will always carry the same epoch number on // the subsequent reconnections. optional uint64 topic_epoch = 11; } enum ProducerAccessMode { Shared = 0; // By default multiple producers can publish on a topic Exclusive = 1; // Require exclusive access for producer. Fail immediately if there's already a producer connected. WaitForExclusive = 2; // Producer creation is pending until it can acquire exclusive access }
这个是作为CommandProduce 请求的成功response
/// Response from CommandProducer message CommandProducerSuccess { // 网络层id required uint64 request_id = 1; // producer 名字 required string producer_name = 2; // The last sequence id that was stored by this producer in the previous session // This will only be meaningful if deduplication has been enabled. optional int64 last_sequence_id = 3 [default = -1]; optional bytes schema_version = 4; // The topic epoch assigned by the broker. This field will only be set if we // were requiring exclusive access when creating the producer. optional uint64 topic_epoch = 5; // 这个应该和上面ProducerAccessMode 相关,后面有机会来介绍这个吧 // If producer is not "ready", the client will avoid to timeout the request // for creating the producer. Instead it will wait indefinitely until it gets // a subsequent `CommandProducerSuccess` with `producer_ready==true`. optional bool producer_ready = 6 [default = true]; }
这个是producer 用来发送消息到服务端用的RPC
可以通过org.apache.pulsar.client.impl.ProducerImpl.sendAsync
这个方法一路追到这个调用的位置,一般消息经过batch,加密,分块等逻辑处理之后,会将消息序列化成这个请求。
具体序列化的格式是下面这个
BaseCommand就是CommandSend
// org.apache.pulsar.common.protocol.Commands private static ByteBufPair serializeCommandSendWithSize(BaseCommand cmd, ChecksumType checksumType, MessageMetadata msgMetadata, ByteBuf payload) { // / Wire format // [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD]
这里面的protocol格式实际只包含了上面的 [CMD] 部分
message CommandSend { required uint64 producer_id = 1; required uint64 sequence_id = 2; optional int32 num_messages = 3 [default = 1]; optional uint64 txnid_least_bits = 4 [default = 0]; optional uint64 txnid_most_bits = 5 [default = 0]; /// Add highest sequence id to support batch message with external sequence id optional uint64 highest_sequence_id = 6 [default = 0]; optional bool is_chunk =7 [default = false]; }
这个是服务端成功处理完消息持久化之后成功的response
message CommandSendReceipt { required uint64 producer_id = 1; // 这个是用来保证顺序的 required uint64 sequence_id = 2; optional MessageIdData message_id = 3; // 这个应该是用来去重的 optional uint64 highest_sequence_id = 4 [default = 0]; } // 这个是返回的写入成功的消息id,这个结构会在其他位置复用 message MessageIdData { required uint64 ledgerId = 1; required uint64 entryId = 2; optional int32 partition = 3 [default = -1]; // 这里是 optional int32 batch_index = 4 [default = -1]; repeated int64 ack_set = 5; optional int32 batch_size = 6; }
这个是CommandSend 异常的response
message CommandSendError { required uint64 producer_id = 1; required uint64 sequence_id = 2; required ServerError error = 3; required string message = 4; }
这个是用来告知服务端我这个consumer当前可以接受消息的数目
服务端会记录一个subscription里面每个consumer当前可以接受消息的数目
分配消息给哪个consumer的时候会按照这个数目来确定consumer当前能否接受消息。
目前了解到的位置是在connectionOpened
的这个方法成功处理Subscription 注册之后会发送一个CommandFlow
请求,来让服务端推送消息。
不过可以想到,如果consumer队列是空闲的状态下都会发送这个消息。
message CommandFlow { required uint64 consumer_id = 1; // Max number of messages to prefetch, in addition // of any number previously specified required uint32 messagePermits = 2; }
这里实际上可能是服务端推消息给consumer,服务端会主动发送这个请求给consumer。(这个逻辑在服务端的 subscription 里的 dispatcher里面)
具体的调用位置在 org.apache.pulsar.broker.service.Consumer#sendMessages
这个方法在往上看一层的话都是org.apache.pulsar.broker.service.Dispatcher
这个类调用的。
这里和上面写入的格式一样这里的Command 实际上是一个RPC的header后面会加上消息的payload。
// Wire format // [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD] // // metadataAndPayload contains from magic-number to the payload included
message CommandMessage { required uint64 consumer_id = 1; // 这里是消息的id required MessageIdData message_id = 2; // 这个消息重发了多少次 optional uint32 redelivery_count = 3 [default = 0]; // 这个消息里面哪些已经被ack了 repeated int64 ack_set = 4; }
这个用来ack成功消费的消息,可以单独ack一条消息,
也可以累积确认(类似kafka)。
这里为了减少RPC的频率,在客户端做了一个batch ack 的优化。
服务端的对应处理一般会更新ManagedCursor
里面保存的数据,将这个ack的结果持久化。
message CommandAck { // ack 类型,是累积确认还是单独确认 enum AckType { Individual = 0; Cumulative = 1; } required uint64 consumer_id = 1; required AckType ack_type = 2; // 这里类型是repeated类型的可以把ack做batch // In case of individual acks, the client can pass a list of message ids repeated MessageIdData message_id = 3; // Acks can contain a flag to indicate the consumer // received an invalid message that got discarded // before being passed on to the application. enum ValidationError { UncompressedSizeCorruption = 0; DecompressionError = 1; ChecksumMismatch = 2; BatchDeSerializeError = 3; DecryptionError = 4; } // 一些异常情况可能也会ack这个消息,这里会记录一些信息 optional ValidationError validation_error = 4; repeated KeyLongValue properties = 5; optional uint64 txnid_least_bits = 6 [default = 0]; optional uint64 txnid_most_bits = 7 [default = 0]; // 网络层请求id optional uint64 request_id = 8; }
这个是consumer告诉服务端哪些消息需要重新被投递的RPC
message CommandRedeliverUnacknowledgedMessages { required uint64 consumer_id = 1; repeated MessageIdData message_ids = 2; }
这个其实是一个公用的response,如果请求没有特殊需要返回的字段的话,几乎可以被所有的请求使用。
这里不像Kafka 每个request和response 都带着一个ApiKey
不会严格一一对应。
message CommandSuccess { required uint64 request_id = 1; optional Schema schema = 2; } message CommandError { required uint64 request_id = 1; required ServerError error = 2; required string message = 3; }
这2个都是空的,主要作用是用来维护tcp连接应用层的keepAlive
org.apache.pulsar.common.protocol.PulsarHandler#handleKeepAliveTimeout
// Commands to probe the state of connection. // When either client or broker doesn't receive commands for certain // amount of time, they will send a Ping probe. message CommandPing { } message CommandPong { }
到此,关于“Apache Pulsar二进制协议怎么实现”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注亿速云网站,小编会继续努力为大家带来更多实用的文章!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。