这里主要分析kafka 客户端实现 (代码分析以perl kafka实现为准)
kafka客户端分为生产者和消费者,生产者发送消息,消费者获取消息.
在kafka协议里客户端通信中用到的最多的四个协议命令是fetch,fetchoffset,send,metadata.这四个分别是获取消息,获取offset,发送消息,获取metadata.剩下的其他协议命令大多都是kafka server内部通信用到的.offsetcommit这个命令在有些语言的client api的实现里给出了接口可以自己提交offset.但是在perl的实现里并没有.
先看看直接producer和consumer的代码
my $request = { ApiKey => $APIKEY_PRODUCE, CorrelationId => $self->{CorrelationId}, ClientId => $self->{ClientId}, RequiredAcks => $self->{RequiredAcks}, Timeout => $self->{Timeout} * 1000, topics => [ { TopicName => $topic, partitions => [ { Partition => $partition, MessageSet => $MessageSet, }, ], }, ], }; foreach my $message ( @$messages ) { push @$MessageSet, { Offset => $PRODUCER_ANY_OFFSET, Key => $key, Value => $message, }; } return $self->{Connection}->receive_response_to_request( $request, $compression_codec );
代码并未完全贴上.核心代码就这一部分.最后一行代码可以看见最终调用connection::receive_response_to_request函数.再上面的部分是设置消息格式.和消息内容的数据结构.
my $request = { ApiKey => $APIKEY_FETCH, CorrelationId => $self->{CorrelationId}, ClientId => $self->{ClientId}, MaxWaitTime => $self->{MaxWaitTime}, MinBytes => $self->{MinBytes}, topics => [ { TopicName => $topic, partitions => [ { Partition => $partition, FetchOffset => $start_offset, MaxBytes => $max_size // $self->{MaxBytes}, }, ], }, ], }; my $response = $self->{Connection}->receive_response_to_request( $request ); 这是consumer的获取消息的核心部分代码.最后同producer一样.代码结构也相似.同样是设置消息数据结构然后发送.只是最后多了代码返回处理的部分.消息返回处理的部分就不再贴上详细说明了.有兴趣自行去cpan上看源代码.
下面看看最核心的函数代码.
sub receive_response_to_request { my ( $self, $request, $compression_codec ) = @_; local $Data::Dumper::Sortkeys = 1 if $self->debug_level; my $api_key = $request->{ApiKey}; //这里获取请求类型,是发送消息,还是获取消息和offset的. # WARNING: The current version of the module limited to the following:# supports queries with only one combination of topic + partition (first and only). my $topic_data = $request->{topics}->[0]; //这些消息具体处理就略过不提了. my $topic_name = $topic_data->{TopicName}; my $partition = $topic_data->{partitions}->[0]->{Partition}; if ( //这里是比较关键的.判断是否有完整的metadata信息.没有metadata信息就通过fetch meta命令获取. !%{ $self->{_metadata} } # the first request || ( !$self->{AutoCreateTopicsEnable} && defined( $topic_name ) && !exists( $self->{_metadata}->{ $topic_name } ) ) ) { //updata_metadata函数就是封装了fetch metadata请求命令发送给kafka 来获取metadata信息.在这个地方处理不同语言里处理逻辑多少有些差别.php-kafka中有两种方式,一种通过这里的这个方法.另一种是通过zookeeper获取meta信息.在使用的时候需要指定zookeeper地址. $self->_update_metadata( $topic_name ) # hash metadata could be updated # FATAL error or $self->_error( $ERROR_CANNOT_GET_METADATA, format_message( "topic = '%s'", $topic_name ) ); } my $encoded_request = $protocol{ $api_key }->{encode}->( $request, $compression_codec ); //这里将消息格式化成网络字节序. my $CorrelationId = $request->{CorrelationId} // _get_CorrelationId; say STDERR sprintf( '[%s] compression_codec = %d, request: %s', scalar( localtime ), $compression_codec // ' ', Data::Dumper->Dump( [ $request ], [ 'request' ] ) ) if $self->debug_level; my $attempts = $self->{SEND_MAX_ATTEMPTS}; my ( $ErrorCode, $partition_data, $server ); ATTEMPTS: while ( $attempts-- ) { //在while里进行发送尝试.java版客户端的三次尝试即是这里同样的逻辑 REQUEST: { $ErrorCode = $ERROR_NO_ERROR; //这里差早topic分区对应的leader,成功则进行leader连接发送请求 if ( defined( my $leader = $self->{_metadata}->{ $topic_name }->{ $partition }->{Leader} ) ) { # hash metadata could be updated unless ( $server = $self->{_leaders}->{ $leader } ) { //没有找到对应leader的server就跳过此次请求尝试,更新metadata并进行下一次尝试 $ErrorCode = $ERROR_LEADER_NOT_FOUND; $self->_remember_nonfatal_error( $ErrorCode, $ERROR{ $ErrorCode }, $server, $topic_name, $partition ); last REQUEST; # go to the next attempt //在这里跳出主逻辑块.进行块之后的动作. } # Send a request to the leader if ( !$self->_connectIO( $server ) ) { //这里连接此topic分区的leader $ErrorCode = $ERROR_CANNOT_BIND; } elsif ( !$self->_sendIO( $server, $encoded_request ) ) { //这里向这个leader发送请求 $ErrorCode = $ERROR_CANNOT_SEND; } if ( $ErrorCode != $ERROR_NO_ERROR ) { //判断动作是否成功 $self->_remember_nonfatal_error( $ErrorCode, $self->{_IO_cache}->{ $server }->{error}, $server, $topic_name, $partition ); last REQUEST; # go to the next attempt } my $response; //这里处理返回情况.如果发送的produce请求并且没有任何response返回.则构建一个空的response返回. if ( $api_key == $APIKEY_PRODUCE && $request->{RequiredAcks} == $NOT_SEND_ANY_RESPONSE ) { # Do not receive a response, self-forming own response $response = { CorrelationId => $CorrelationId, topics => [ { TopicName => $topic_name, partitions => [ { Partition => $partition, ErrorCode => 0, Offset => $BAD_OFFSET, }, ], }, ], }; } else { //这里获取response.并从网络字节序转换成字符格式. my $encoded_response_ref; unless ( $encoded_response_ref = $self->_receiveIO( $server ) ) { if ( $api_key == $APIKEY_PRODUCE ) {# WARNING: Unfortunately, the sent package (one or more messages) does not have a unique identifier# and there is no way to verify the delivery of data $ErrorCode = $ERROR_SEND_NO_ACK; # Should not be allowed to re-send data on the next attempt # FATAL error $self->_error( $ErrorCode, $self->{_IO_cache}->{ $server }->{error} ); } else { $ErrorCode = $ERROR_CANNOT_RECV; $self->_remember_nonfatal_error( $ErrorCode, $self->{_IO_cache}->{ $server }->{error}, $server, $topic_name, $partition ); last REQUEST; # go to the next attempt } } if ( length( $$encoded_response_ref ) > 4 ) { # MessageSize => int32 $response = $protocol{ $api_key }->{decode}->( $encoded_response_ref ); say STDERR sprintf( '[%s] response: %s', scalar( localtime ), Data::Dumper->Dump( [ $response ], [ 'response' ] ) ) if $self->debug_level; } else { $self->_error( $ERROR_RESPONSEMESSAGE_NOT_RECEIVED ); } } $response->{CorrelationId} == $CorrelationId # FATAL error or $self->_error( $ERROR_MISMATCH_CORRELATIONID ); $topic_data = $response->{topics}->[0]; $partition_data = $topic_data->{ $api_key == $APIKEY_OFFSET ? 'PartitionOffsets' : 'partitions' }->[0]; if ( ( $ErrorCode = $partition_data->{ErrorCode} ) == $ERROR_NO_ERROR ) { return $response; } elsif ( exists $RETRY_ON_ERRORS{ $ErrorCode } ) { $self->_remember_nonfatal_error( $ErrorCode, $ERROR{ $ErrorCode }, $server, $topic_name, $partition ); last REQUEST; # go to the next attempt } else { # FATAL error $self->_error( $ErrorCode, format_message( "topic = '%s', partition = %s", $topic_name, $partition ) ); } } } # Expect to possible changes in the situation, such as restoration of connection say STDERR sprintf( '[%s] sleeping for %d ms before making request attempt #%d (%s)', scalar( localtime ), $self->{RETRY_BACKOFF}, $self->{SEND_MAX_ATTEMPTS} - $attempts + 1, $ErrorCode == $ERROR_NO_ERROR ? 'refreshing metadata' : "ErrorCode ${ErrorCode}", ) if $self->debug_level; sleep $self->{RETRY_BACKOFF} / 1000; $self->_update_metadata( $topic_name ) //最重要的逻辑在这里.可以看见上面失败则跳出REQUEST块,直接到这里执行更新动作.更新完之后再进行下一次尝试.这个逻辑应对着topic 分区的leader动态切换的.现有leader死了,切换到其他的leader上来.客户端能对此作出应对. # FATAL error or $self->_error( $ErrorCode || $ERROR_CANNOT_GET_METADATA, format_message( "topic = '%s', partition = %s", $topic_name, $partition ) ); } # FATAL error if ( $ErrorCode ) { $self->_error( $ErrorCode, format_message( "topic = '%s'%s", $topic_data->{TopicName}, $partition_data ? ", partition = ".$partition_data->{Partition} : q{} ) ); } else { $self->_error( $ERROR_UNKNOWN_TOPIC_OR_PARTITION, format_message( "topic = '%s', partition = %s", $topic_name, $partition ) ); } return;} 上面主要分析核心逻辑实现.可以发现:
consumer在消费的时候并没有手动提交过offset.也未设置groupId相关的配置,所以在消费的时候server其实并不是强制按group消费的,也不自动记录对应offset.只是按提交的offset返回对应的消息和下一个offset值而已.所以在kafka按组消费的功能其实是有各个客户端api实现的.在新版java的api中可以看见有autoCommitOffset的方法.在老版java api实现里也有autocommit的线程在替用户提交groupId与offset的记录.
producer和consumer的request里均需要指定topic分区.所以实际上在真正的api底层是没有对topic分区做负载的.一些具有负载功能的其他语言的api均由客户端内部实现.并非kafka server提供的功能.