这里主要分析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 // '<undef>',
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提供的功能.
