Configurações de Corretores
| Propriedade | Inadimplência | Descrição | | broker.id | | Cada corretor pode ser identificado com um ID inteiro único e não negativo; Esse id pode ser usado como o "nome" do corretor, e sua existência permite que o corretor migre para outro host/porta sem confundir os consumidores. Você pode escolher qualquer número que quiser como ID, desde que o ID seja único. | | log.dirs | /tmp/kafka-logs | O caminho onde Kafka armazena dados. Esse caminho não é único, pode ser múltiplo, e os caminhos só precisam ser separados por vírgulas; Sempre que uma nova partição é criada, ela é escolhida para fazê-la sob o caminho que contém o menor número de partições. | | Porto | 6667 | Servidor aceita a porta à qual o cliente se conecta | | zookeeper.connect | zero | O formato da string de conexão ZooKeeper é: hostname:port, onde hostname e port são o host e o port de um nó no cluster ZooKeeper, respectivamente. Para se conectar a outros nós do ZooKeeper quando um host cai do ar, você pode criar múltiplos hosts da seguinte forma:
hostname1:port1, hostname2:port2, hostname3:port3.
O ZooKeeper permite adicionar um caminho "chroot" para armazenar todos os dados kafka no cluster sob um caminho específico. Quando múltiplos clusters Kafka ou outras aplicações usam o mesmo cluster ZooKeeper, você pode usar esse método para definir o caminho de armazenamento de dados. Isso pode ser implementado formatando a string de conexão assim:
Nome do Host1:Porta1,Nome do Host2:Porta2,Nome do Host3:Porta3/chroot/path
Essa configuração armazena todos os dados do cluster kafka sob o caminho /chroot/path. Note que, antes de iniciar o corretor, você deve criar esse caminho, e os consumidores devem usar o mesmo formato de conexão. | | message.max.bytes | 1000000 | O tamanho máximo de mensagens que um servidor pode receber. É importante que as configurações do consumidor e do produtor em relação a essa propriedade estejam sincronizadas, caso contrário a mensagem publicada pelo produtor é grande demais para o consumidor. | | num.network.threads | 3 | O número de threads de rede usados pelo servidor para processar solicitações de rede; Geralmente, você não precisa mudar essa função. | | num.io.threads | 8 | O número de threads de E/S usados pelo servidor para processar requisições; O número de threads deve ser pelo menos igual ao número de discos rígidos. | | Contexto.threads | 4 | O número de threads usados para processamento em segundo plano, como exclusão de arquivos; Você não precisa mudar essa propriedade. | | queued.max.solicitações | 500 | O número máximo de solicitações que podem ser enfileiradas para uma thread de E/S processar antes que uma thread de rede pare de ler novas requisições. | | host.name | zero | nome de host do corretor; Se o nome de host já estiver definido, o corretor só será vinculado a esse endereço; Se não houver configuração, ele será vinculado a todas as interfaces e publicará uma cópia no ZK | | advertised.host.name | zero | Se definido, ele é enviado para produtores, consumidores e outros corretores como o nome host do corretor | | anunciado.port | zero | Este porto é concedido a produtores, consumidores e outros corretores, e é usado para estabelecer uma conexão; Só precisa ser definido se a porta real e a porta que o servidor precisa vincular forem diferentes. | | socket.envio.buffer.bytes | 100 * 1024 | SO_SNDBUFF Tamanho do cache, que o servidor usa para fazer conexões de socket | | socket.receber.buffer.bytes | 100 * 1024 | SO_RCVBUFF tamanho do cache, que é usado pelo servidor para se conectar aos sockets | | socket.request.max.bytes | 100 * 1024 * 1024 | O tamanho máximo permitido pelo servidor; Isso evitará overflows de servidores, que devem ser menores que o tamanho do heap Java | | num.partições | 1 | Se o número de partições não for indicado ao criar um tópico, esse número será o número padrão de partições sob o tópico. | | log.segment.bytes | 1014*1024*1024 | Os logs da partição tema são armazenados em muitos arquivos em um determinado diretório, que dividem os logs da partição em segmentos; Esse atributo é o tamanho máximo de cada arquivo; Quando as dimensões atingem esse valor, um novo arquivo é criado. Essa configuração pode ser superada pela base de cada tema.
Visão O login do hiperlink está visível. | | log.roll.hours | 24 * 7 | Mesmo que o arquivo não alcance log.segment.bytes, um novo arquivo é criado sempre que o momento de criação do arquivo atingir essa propriedade. Essa configuração também pode ser sobrescrevida por configurações em nível temático;
VistaO login do hiperlink está visível. | | log.cleanup.policy | delete | | | log.retention.minutes e log.retention.hours | 7 dias | O tempo em que cada arquivo de log foi salvo antes de ser excluído. O tempo padrão de economia de dados é o mesmo para todos os tópicos.
log.retention.minutes e log.retention.bytes são ambos usados para definir a exclusão de arquivos de log, independentemente de qual propriedade tenha sido superada.
Essa configuração de propriedade pode ser anulada quando o tema está basicamente definido.
VistaO login do hiperlink está visível. | | log.retention.bytes | -1 | A quantidade total de dados salvos por cada partição sob cada tópico; Note que esse é o limite superior por partição, então esse número multiplicado pelo número de partições é a quantidade total de dados armazenados por tópico. Também observe: Se tanto log.retention.hours quanto log.retention.bytes estiverem definidos, ultrapassar qualquer um dos limites fará com que um arquivo de segmento seja excluído.
Note que essa configuração pode ser sobreposta por cada tópico.
VistaO login do hiperlink está visível. | | log.retention.check.interval.ms | 5 minutos | Verifique o intervalo entre os arquivos segmentados em log para determinar se os atributos do arquivo atendem aos requisitos de exclusão. | | log.cleaner.enable | false | Quando essa propriedade é definida como falsa, ela será excluída assim que o log for armazenado por um tempo ou tamanho máximo. Se definido como verdadeiro, isso acontecerá quando o atributo de salvamento atingir o limite superiorO login do hiperlink está visível.。 | | log.cleaner.threads | 1 | O número de threads realizando compressão logarítmica | | log.cleaner.io.max.bytes.per.segundo | Nenhum | O número máximo de I/Os que um limpador de troncos pode ter ao realizar uma compactação de logs. Essa configuração limita o limpador para evitar interferir nos serviços de solicitação ativos. | | log.cleaner.io.buffer.size | 500*1024*1024 | O Limpador de Logs indexa os logs durante o processo de limpeza e reduz o tamanho do cache utilizado. É melhor configurá-lo grande para fornecer memória suficiente. | | log.cleaner.io.buffer.load.factor | 512*1024 | O tamanho do bloco de I/O necessário para a limpeza de troncos. Você não precisa mudar essa configuração. | | log.cleaner.io.buffer.load.factor | 0.9 | fator de carga da tabela hash usada na limpeza de logs; Você não precisa mudar essa opção. | | log.cleaner.backoff.ms | 15000 | O intervalo de tempo em que o registro é limpo é realizado | | log.cleaner.min.cleanable.ratio | 0.5 | Essa configuração controla com que frequência o compactador de troncos tenta limpar os troncos (assumidoO login do hiperlink está visível.está aberto). Por padrão, evite limpar mais de 50% dos troncos. Essa razão está ligada ao espaço máximo consumido pelo log de backup (50% dos logs são comprimidos a 50%). Uma taxa maior significa menos desperdício e mais espaço pode ser liberado de forma mais eficiente. Essa configuração pode ser sobreposta em cada configuração de tópico.
VistaO login do hiperlink está visível.。 | | log.cleaner.delete.retention.ms | 1 dia | tempo de armazenamento; O tempo máximo para manter os registros comprimidos; Também é o tempo máximo para o cliente consumir mensagens, e a diferença entre log.retention.minutes é que um controla os dados não comprimidos e o outro controla os dados comprimidos, que serão sobrescritos pelo tempo especificado quando o tópico é criado. | | log.index.size.max.bytes | 10*1024*1024 | O tamanho máximo por segmento de troncado. Note que, se o tamanho logaritário atingir esse valor, um novo segmento logarítmico precisa ser gerado mesmo que o tamanho não exceda o limite de log.segment.bytes. | | log.index.interval.bytes | 4096 | Ao realizar uma busca, você precisa escanear o deslocamento mais próximo com uma certa quantidade de espaço; quanto maior a configuração, melhor, geralmente use o valor padrão | | log.flush.interval.mensagens | Long.MaxValue | arquivo de log "sincroniza" com o disco antes de acumular mensagens. Como as operações de IO do disco são uma operação lenta, mas também um meio necessário de "confiabilidade dos dados", verifique se o intervalo de tempo para curar no disco rígido é necessário. Se esse valor for muito grande, levará a um tempo de "sincronização" muito longo (bloqueio de IO); se esse valor for muito pequeno, levará a um longo período de "fsync" (bloqueio de IO); se esse valor for muito pequeno, levará a um grande número de tempos de "sincronização", o que significa que a solicitação geral do cliente terá certo atraso, e a falha do servidor físico levará à perda de mensagens sem fsync. | | log.flush.scheduler.interval.ms | Long.MaxValue | Verifique se intervalos fsync são necessários | | log.flush.interval.ms | Long.MaxValue | Esse número é usado para controlar o intervalo de tempo do "fsync"; se o número de mensagens nunca atingir o número de mensagens solidificadas no disco, mas o intervalo de tempo desde a última sincronização do disco atingir o limiar, a sincronização do disco também será acionada. | | log.delete.delay.ms | 60000 | O tempo de retenção após o arquivo ser limpado no índice geralmente não precisa ser modificado | | auto.create.topics.enable | true | Se permitir a criação automática de tópicos. Se for verdade, isso automaticamente criará um tópico que não existe quando produzir ou buscar não existe. Caso contrário, você precisa usar a linha de comando para criar um tópico | | controller.socket.timeout.ms | 30000 | O tempo de timeout do soquete quando o controlador de gerenciamento de partições realiza um backup. | | controlador.mensagem.fila.tamanho | Int.MaxValue | controlador-para-corretor-channles | | fator padrão.replicação | 1 | O número padrão de cópias de backup refere-se apenas a tópicos criados automaticamente | | replica.lag.time.max.ms | 10000 | Se um seguidor não enviar um pedido de busca dentro desse tempo, o líder removerá o seguidor do ISR e considerará que ele foi enforcado | | replica.lag.max.mensagens | 4000 | Se uma réplica tiver mais de esse número de sem backup, o líder removerá o seguidor e considerará que ele foi pendurado | | replica.socket.timeout.ms | 30*1000 | Tempo de timeout do líder para requisições de rede socket ao fazer backup de dados | | replica.socket.receive.buffer.bytes | 64*1024 | Buffer de recepção de soquete ao enviar uma requisição de rede ao líder durante o backup | | replica.fetch.max.bytes | 1024*1024 | O valor máximo de cada busca no momento do backup | | replica.fetch.min.bytes | 500 | O tempo máximo de espera para que os dados cheguem ao líder quando este faz uma solicitação de backup | | replica.fetch.min.bytes | 1 | O menor tamanho da resposta após cada busca ao fazer backup | | num.replica.fetchers | 1 | O número de threads que fazem backup dos dados do líder | | replica.high.watermark.checkpoint.interval.ms | 5000 | Cada réplica verifica com que frequência o nível mais alto de água é curado | | fetch.purgatory.purge.interval.requests | 1000 | Busca O intervalo de purga | | producer.purgatory.purge.interval.requests | 1000 | Produtor solicita um intervalo de purga | | zookeeper.session.timeout.ms | 6000 | Tempo para a sessão do tratador. | | zookeeper.connection.timeout.ms | 6000 | O tempo máximo que o cliente espera para estabelecer uma conexão com o tratador de zoológico | | zookeeper.sync.time.ms | 2000 | Seguidor ZK fica atrás do líder ZK por muito tempo | | controlled.shutdown.enable | true | Se é possível controlar o encerramento do corretor. Se possível, o corretor poderá transferir todos os líderes para outros corretores antes do fechamento. Isso reduz a indisponibilidade durante o processo de desligamento. | | controlled.shutdown.max.retentativas | 3 | O número de comandos que podem executar com sucesso um desligamento antes de realizar um desligamento incompleto. | | controlled.shutdown.retry.backoff.ms | 5000 | Tempo de descanso entre os desligamentos | | auto.leader.rebalance.enable | true | Se isso for verdade, o controlador automaticamente equilibrará a liderança dos corretores sobre as partições | | líder.desequilíbrio.per.corretor.percentual | 10 | A razão máxima de desequilíbrio permitida por cada corretor | | líder.desequilíbrio.cheque.intervalo.segundos | 300 | Verifique a frequência do desequilíbrio do líder | | offset.metadata.max.bytes | 4096 | Permite que os clientes salvem o máximo de seus deslocamentos | | max.connections.per.ip | Int.MaxValue | O número máximo de conexões por corretor pode ser feito para cada endereço IP | | max.connections.per.ip.overrides | | A cobertura máxima da conexão padrão por IP ou nome de host | | connections.max.idle.ms | 600000 | Limite de tempo para conexões vazias | | Log.roll.jitter. {ms,hours} | 0 | O número máximo de jitters abstraídos de logRollTimeMillis | | num.recovery.threads.per.data.dir | 1 | O número de threads que cada diretório de dados usa para registrar a recuperação | | sujo.líder.eleição.permitir | true | Indica se é possível usar a configuração de não-réplicas no ISR como líder | | delete.topic.enable | false | Possível deletar tópicos | | offsets.topic.num.partitions | 50 | O número de partições para o tópico de commit de deslocamento. Como mudar isso após a implantação atualmente não é suportado, recomendamos usar uma configuração mais alta para produção (por exemplo, 100-200). | | offsets.topic.retention.minutes | 1440 | Deslocamentos que existem há mais tempo que esse limite de tempo serão marcados como exclusão pendente | | offsets.retention.check.interval.ms | 600000 | O gerente de offset verifica a frequência dos deslocamentos obsoletos | | offsets.topic.replication.factor | 3 | O número de cópias de backup do deslocamento do tema. Recomenda-se definir números mais altos para garantir maior disponibilidade | | offset.topic.segment.bytes | 104857600 | Muda de tema. | | offsets.load.buffer.size | 5242880 | Essa configuração está relacionada ao tamanho do lote e é usada ao ler do segmento de deslocamentos. | | offsets.commit.required.acks | -1 | O número de confirmações precisa ser definido antes que o commit offset seja aceitável, e geralmente não precisa ser alterado |
| Propriedade | Inadimplência | Propriedade Padrão do Servidor | Descrição | | cleanup.policy | delete | log.cleanup.policy | Ou "delete" ou "compactar"; Essa cadeia indica como utilizar a parte antiga do tronco; O método padrão ("delete") descartará a peça antiga quando o limite de tempo ou tamanho de reciclagem for atingido. "compactar" vai comprimir os troncos | | delete.retention.ms | 86400000 (24 horas) | log.cleaner.delete.retention.ms | A diferença entre log.retention.minutes é que um controla dados não comprimidos e o outro controla dados comprimidos. Essa configuração pode ser sobreposta pelos parâmetros de fixação quando o tópico é criado | | flush.mensagens | Nenhum | log.flush.interval.mensagens | Essa configuração especifica um intervalo de tempo para forçar logs fsync. Por exemplo, se essa opção estiver definida como 1, então fsync é necessário após cada mensagem, e se definido para 5, fsync é necessário para cada 5 mensagens. Em geral, recomenda-se que você não defina esse valor. A definição desse parâmetro requer o compromisso necessário entre "confiabilidade dos dados" e "desempenho". Se esse valor for muito grande, causará um longo tempo para "fsync" a cada vez (bloqueio de IO), e se esse valor for muito pequeno, gerará um grande número de "fsync", o que também significa que há um certo atraso na solicitação geral do cliente. Falha física do servidor fará com que as mensagens sejam perdidas sem fsync. | | flush.ms | Nenhum | log.flush.interval.ms | Essa configuração é usada para fixar o intervalo de tempo entre forçar logs fsync para o disco; Por exemplo, se configurado para 1000, então o fsync é necessário a cada 1000ms. Essa opção geralmente não é recomendada | | index.intervalo.bytes | 4096 | log.index.interval.bytes | A configuração padrão garante que adicionemos um índice à mensagem a cada 4096 bytes, e mais índices tornam a mensagem de leitura mais próxima, mas o tamanho do índice será aumentado. Essa opção geralmente não é obrigatória | | max.message.bytes | 1000000 | max.message.bytes | O tamanho máximo da mensagem de anexação kafka. Note que, se você aumentar esse tamanho, também deve aumentar o tamanho do seu consumidor para que ele possa buscar mensagens até esses tamanhos máximos. | | Proporção.min.limpável.suja | 0.5 | Proporção.min.limpável.suja | Essa configuração controla com que frequência o compressor de logs tenta purgar os logs. Por padrão, toras com taxa de compressão superior a 50% serão evitadas. Essa razão evita o maior desperdício de espaço | | min.insync.replicas | 1 | min.insync.replicas | Quando o produtor está configurado para request.required.acks em -1, min.insync.replicas especifica o número mínimo de réplicas (cada gravação de repica deve ser bem-sucedida), e se esse número não for alcançado, o produtor produzirá uma exceção. | | retention.bytes | Nenhum | log.retention.bytes | Se você usar a política de retenção "delete", essa configuração se refere ao tamanho máximo que o log pode atingir antes de ser excluído. Por padrão, não há limite de tamanho, apenas um limite de tempo | | retention.ms | 7 dias | log.retention.minutes | Se você usar a política de retenção de "exclusão", essa configuração se refere ao momento em que o log foi salvo antes do log de exclusão. | | segment.bytes | 1GB | log.segment.bytes | Em Kafka, os troncos são armazenados em blocos, e essa configuração refere-se ao tamanho dos troncos divididos em blocos | | segment.index.bytes | 10MB | log.index.size.max.bytes | Essa configuração é aproximadamente o tamanho do arquivo de índice mapeado entre os deslocamentos e as localizações dos arquivos; Essa configuração geralmente não precisa ser modificada | | segment.ms | 7 dias | log.roll.hours | Mesmo que o arquivo de blocos de log não atinja o tamanho necessário para ser deletado ou comprimido, uma vez que o tempo de log atinja esse limite superior, um novo arquivo de bloco de log será forçado | | segment.jitter.ms | 0 | Log.roll.jitter. {ms,hours} | O jitter máximo a subtrair de logRollTimeMillis. |
Configurações de Consumidor
| Propriedade | Inadimplência | Descrição | | group.id | | Uma string que identifica de forma única o grupo no qual o processo consumidor está localizado e, se o mesmo ID de grupo for definido, significa que esses processos pertencem ao mesmo grupo consumidor | | zookeeper.connect | | Especifique a string da conexão Zookeeper, o formato é hostname:port, aqui host e porta são tanto host quanto porta do servidor Zookeeper, para evitar perder contato após uma máquina Zookeeper cair, você pode especificar múltiplos hostname:ports, usando vírgulas como separação:
nome do host1:porta1,nome do host2:porto2,nome do host3:porto3
Você pode adicionar o caminho chroot do ZooKeeper à string de conexão do ZooKeeper, que é usada para armazenar seus próprios dados, de uma forma:
Nome do Host1:Porta1,Nome do Host2:Porta2,Nome do Host3:Porta3/chroot/path | | consumer.id | zero | Não é necessário configurar e geralmente é gerado automaticamente | | socket.timeout.ms | 30*100 | Limites de tempo para solicitações de rede. O limite real de tempo é max.fetch.wait+socket.timeout.ms | | socket.receber.buffer.bytes | 64*1024 | Socket é usado para receber o tamanho do cache das requisições de rede | | fetch.message.max.bytes | 1024*1024 | O número máximo de bytes por mensagem de busca por requisição de busca. Esses bytes serão supervisionados na memória usada para cada partição, então essa configuração controlará a quantidade de memória usada pelo consumidor. O tamanho da requisição de busca deve ser pelo menos igual ao tamanho máximo permitido pelo servidor, caso contrário, o tamanho da mensagem que o produtor pode enviar é maior do que o tamanho que o consumidor pode consumir. | | num.consumidor.buscadores | 1 | O número de threads de busca usadas para dados de busca | | auto.commit.enable | true | Se for verdadeiro, o deslocamento da mensagem buscada pelo consumidor será automaticamente sincronizado com o tratador. Esse deslocamento de compromisso será usado pelo novo consumidor quando o processo travar | | auto.commit.interval.ms | 60*1000 | A frequência com que o consumidor envia o deslocamento ao tratador é em segundos | | queued.max.message.chunks | 2 | O número máximo de mensagens usadas em cache para consumo. Cada bloco deve ser igual a fetch.message.max.bytes | | rebalance.max.retentativas | 4 | Quando um novo consumidor é adicionado a um grupo de consumidores, o conjunto tenta reequilibrar o número de partições alocadas a cada consumidor. Se a cobrança do consumidor mudar, esse rebalanceamento falha e se reentrifica quando a alocação está sendo executada | | fetch.min.bytes | 1 | O número mínimo de bytes que o servidor deve devolver a cada requisição de busca. Se não forem retornados dados suficientes, a solicitação espera até que haja dados suficientes. | | fetch.wait.max.ms | 100 | Se não houver dados suficientes para satisfazer o fetch.min.bytes, essa configuração refere-se ao tempo máximo que o servidor bloqueará antes de responder a uma solicitação de busca. | | rebalance.backoff.ms | 2000 | Tempo de recuo antes de tentar reabrir novamente | | refresh.leader.backoff.ms | 200 | Há um tempo de recuo para esperar antes de tentar determinar se o líder de uma partição perdeu sua liderança | | auto.offset.reset | maior | Se não houver deslocamento inicializado no zookeeper, se o deslocamento for uma resposta ao seguinte valor:
menor: Reset automático do deslocamento para o menor deslocamento
maior: Reset automático deslocado para o deslocamento do maior
Qualquer outra coisa: Faz uma exceção para o consumidor | | consumer.timeout.ms | -1 | Se nenhuma mensagem estiver disponível, mesmo após esperar um tempo específico, uma exceção de tempo limite é lançada | | exclude.tópicos internos | true | Se expor mensagens de tópicos internos para os consumidores | | paritition.assignment.strategy | Distribuição | Selecione a política para atribuir partições ao fluxo do consumidor, opcionalmente range, roundrobin | | client.id | Valor do ID do grupo | é uma string específica para o usuário que ajuda a rastrear chamadas em cada requisição. Ele deve confirmar logicamente a aplicação que gerou a solicitação | | zookeeper.session.timeout.ms | 6000 | Limites de tempo para sessões com tratadores. Se o consumidor não enviar uma mensagem de batimentos cardíacos ao tratador durante esse período, considera-se que ele está desligado e uma reação será gerada | | zookeeper.connection.timeout.ms | 6000 | O tempo máximo de espera para um cliente estabelecer uma conexão com o Tratador de Zoológico | | zookeeper.sync.time.ms | 2000 | Seguidores ZK podem ficar atrás do líder ZK por um tempo máximo | | offsets.storage | tratador de zoológico | Lugares usados para armazenar compensações: zoológico ou kafka | | offset.channel.backoff.ms | 1000 | O tempo de backoff para reconectar ao canal de offsets ou tentar novamente a solicitação de fetch/commit do offset que falhou | | offsets.channel.socket.timeout.ms | 10000 | O limite de tempo de espera do socket para a resposta à resposta de busca ou de requisição de commit ao ler o offset. Esse limite de tempo é usado pela solicitação consumerMetadata para solicitar gerenciamento de offset | | offsets.commit.max.retentativas | 5 | O número de vezes que o commit offset foi tentado novamente. Essa retentativa é aplicada apenas para deslocar commits entre desligamentos. Ele | | dual.commit.enabled | true | Se você usar "kafka" como offsets.storage, pode comprometer offset para zookeeper duas vezes (e uma para kafka). Isso é essencial ao migrar do armazenamento offset baseado em tratador para armazenamento offset baseado em kafka. Para qualquer grupo de consumidores, é uma recomendação segura desativar essa opção quando a migração estiver completa | | partição.atribuição.estratégia | Distribuição | Escolha entre as políticas de "range" e "roundrobin" como política para atribuir partições aos fluxos de dados do consumidor; O alocador de partições circulares aloca todas as partições disponíveis, bem como todas as threads de consumo disponíveis. Ele atribuirá o loop de partição à thread de consumidor. Se todas as instâncias de consumo forem subscritas a um determinado, as partições são divididas em distribuições determinísticas. A estratégia de alocação round-robin só é possível se as seguintes condições forem atendidas: (1) Cada tópico tiver o mesmo número de fluxos de dados por força do consumidor. (2) A coleção de tópicos inscritos é determinada para cada instância de consumidor no grupo de consumidores. |
Configurações do Produtor
| Propriedade | Inadimplência | Descrição | | metadata.broker.list | | Saquear de forma independente. O produtor é usado apenas para obter metadados (tópicos, partições, réplicas). A conexão do soquete para enviar os dados reais será estabelecida com base nos metadados retornados. O formato é:
host1:porto1,anfitrião2:porta2
Essa lista pode ser uma sublista de corretores ou um VIP apontando para corretores | | request.required.acks | 0 | Essa configuração é um valor de confirmação que indica quando uma solicitação de produção é considerada concluída. Em particular, quantos outros corretores devem ter enviado dados para seus registros e confirmado essa informação ao seu líder. Valores típicos incluem:
0: Indica que o produtor nunca espera confirmação do corretor (mesmo comportamento do 0,7). Essa opção oferece a menor latência, mas ao mesmo tempo o maior risco (porque os dados são perdidos quando o servidor cai).
1: Indica que a réplica líder recebeu a confirmação dos dados. Essa opção tem baixa latência e garante que o servidor confirme que foi recebida.
-1: O produtor recebe confirmação de que todas as réplicas sincronizadas receberam dados. A latência é a maior, porém esse método não elimina completamente o risco de perda de mensagens, pois o número de réplicas sincronizadas pode ser 1. Se você quiser garantir que algumas réplicas recebam dados, deve definir a opção min.insync.replicas nas configurações de nível de tópico. Leia a documentação do projeto para uma discussão mais aprofundada. | | request.timeout.ms | 10000 | O corretor faz o possível para implementar o requisito request.required.acks, caso contrário, um erro será enviado ao cliente | | produtor.type | Sincronização | Essa opção fixa se a mensagem é enviada assíncrona em um thread em segundo plano. Valores corretos:
(1) assíncrono: Envio assíncrono
(2) sincronização: Envio sincronizado
Ao definir o produtor como assíncrono, podemos processar requisições em lotes (o que é bom para maior throughput), mas isso também cria a possibilidade de a máquina cliente perder dados não enviados | | serializer.class | kafka.serializer.DefaultEncoder | A categoria de serialização da mensagem. O codificador padrão insere um byte[] e retorna o mesmo byte[] | | key.serializer.class | | Classe de serialização para palavras-chave. Se isso não for indicado, o padrão é corresponder à mensagem | | partitioner.class | kafka.producer.DefaultPartitioner | classe partitioner para dividir mensagens entre subtópicos. O particionador padrão é baseado na tabela hash da chave | | compression.codec | nenhum | Esse parâmetro pode definir o codec para comprimir dados, que pode ser selecionado como "nenhum", "gzip", "snappy". | | tópicos comprimidos | zero | Esse parâmetro pode ser usado para determinar se certos tópicos são comprimidos. Se o codec comprimido for um codec diferente do NoCompressCodec, esses codecs são aplicados aos dados especificados dos tópicos. Se a lista de tópicos comprimidos estiver vazia, aplique o codec comprimido específico a todos os tópicos. Se o codec comprimido for NoCompressionCodec, a compressão não está disponível para todos os tópicos. | | message.send.max.retentativas | 3 | Esse parâmetro fará com que o produtor tente automaticamente resolicitações de envio falhadas. Esse parâmetro fixa o número de tentativas. Nota: Definir um valor diferente de 0 fará com que certos erros de rede se repitam: causar um envio e uma perda de confirmação | | retry.backoff.ms | 100 | Antes de cada tentativa novamente, o produtor atualiza os metadados do tópico relevante para verificar se o novo líder é designado. Como a seleção do líder leva um pouco de tempo, essa opção especifica quanto tempo o produtor precisa esperar antes de atualizar os metadados. | | topic.metadata.refresh.interval.ms | 600*1000 | O produtor geralmente atualiza os metadados do tópico em alguns cenários de falha (partição faltante, líder indisponível, etc.). Ele vai seguir um ciclo regular. Se você definir como um valor negativo, os metadados só serão atualizados se falhar. Se configurado para 0, os metadados são atualizados após cada mensagem enviada (essa opção não é recomendada, pois o sistema consome demais). Importante: As atualizações só ocorrem após a mensagem ser enviada, então, se o produtor nunca enviar a mensagem, os metadados nunca são atualizados. | | queue.buffering.max.ms | 5000 | O intervalo máximo de tempo no qual o usuário armazena em cache os dados quando o modo assíncrono é aplicado. Por exemplo, se a mensagem estiver definida para 100, mensagens dentro de 100ms serão processadas em lotes. Isso vai melhorar a taxa de transferência, mas aumentar a latência devido ao cache. | | queue.buffering.max.mensagens | 10000 | Ao usar o modo assíncrono, o número máximo de mensagens não enviadas que podem ser armazenadas em cache na fila antes do produtor deve ser bloqueado ou os dados devem ser perdidos | | batch.num.mensagens | 200 | Ao usar o modo assíncrono, você pode processar em lote o número máximo de mensagens. Ou o número de mensagens chegou online ou queue.buffer.max.ms chegou, e o produtor vai processar | | enviar.buffer.bytes | 100*1024 | Tamanho do cache de gravação do soquete | | client.id | “” | Esse id de cliente é uma string específica do usuário que é incluída em cada solicitação para rastrear a chamada, e ele deve ser logicamente capaz de confirmar que a aplicação fez a solicitação. |
Configurações do Produtor
| Nome | Tipo | Inadimplência | Importância | Descrição | | boostrap.servers | Lista | | alto | Grupo host/port para estabelecer uma conexão com o cluster kafka. Os dados serão carregados de forma uniforme em todos os servidores, independentemente de qual servidor seja designado para bootstrapping. Essa lista afeta apenas os hosts inicializados (que é usada para descobrir todos os servidores). Este formato de lista:
host1:port1,host2:port2,...
Como esses servidores são usados apenas para inicializar conexões e descobrir todas as pertinências do cluster (que podem mudar dinamicamente), essa lista não precisa conter todos os servidores (você pode querer mais de um servidor, embora nesse caso, um possa estar fora do ar). Se nenhum servidor aparecer nessa lista, o envio de dados falhará até que a lista esteja disponível. | | acks | String | 1 | alto | O produtor precisa de um sinal do servidor para confirmar o recebimento após receber os dados, e essa configuração se refere a quantos desses sinais de confirmação o produtor precisa. Essa configuração na verdade representa a disponibilidade de backups de dados. As seguintes configurações são opções comuns:
(1) acks=0: Definido para 0 significa que o produtor não precisa esperar por nenhuma confirmação das informações recebidas. A réplica será imediatamente adicionada ao buffer do soquete e considerada enviada. Não há garantia de que o servidor tenha recebido os dados com sucesso nesse caso, e a tentativa de re-configuração não funcionará (porque o cliente não sabe se falhou) e o deslocamento do feedback sempre será definido para -1.
(2) acks=1: Isso significa que pelo menos espere que o líder escreva os dados com sucesso no log local, mas não que todos os seguidores escrevam com sucesso. Nesse caso, se o seguidor não conseguir fazer backup dos dados e o líder desligar novamente, a mensagem será perdida.
(3) acks=todos: Isso significa que o líder precisa esperar todos os backups para gravar logs com sucesso, e essa estratégia garantirá que os dados não sejam perdidos enquanto um backup sobreviver. Essa é a maior garantia.
(4) Outras configurações, como acks=2, também são possíveis, que exigirão um determinado número de acks, mas essa estratégia geralmente é raramente usada. | | buffer.memory | Longas | 33554432 | alto | O produtor pode ser usado para armazenar em cache o tamanho da memória dos dados. Se os dados forem gerados mais rápido do que enviados ao corretor, o produtor bloqueará ou lançará uma exceção, indicada por "block.on.buffer.full".
Essa configuração estará relacionada à memória total que o produtor pode usar, mas não é um limite rígido, pois nem toda a memória usada pelo produtor é usada para cache. Alguma memória adicional é usada para compressão (se for introduzida) e parte é usada para solicitações de manutenção. | | compression.type | String | nenhum | alto | produtor é o tipo de compressão usada para comprimir dados. O padrão é não comprimido. Os valores corretos das opções são nenhum, gzip, snappy.
A compressão é melhor usada para processamento em lote; quanto mais mensagens são processadas em lotes, melhor será o desempenho da compressão. | | Retentativas | int | 0 | alto | Definir um valor maior que 0 fará com que o cliente reenvie qualquer dado assim que esses dados falharem. Note que essas tentativas não são diferentes daquelas em que o cliente recebe um erro de envio. Permite que tentativas de reação possam potencialmente alterar a ordem dos dados; se ambos os registros de mensagem forem enviados para a mesma partição, a primeira mensagem falhar, a segunda aparecer antes da primeira. | | batch.size | int | 16384 | Média | O produtor tentará agrupar os registros de mensagens para reduzir o número de solicitações. Isso vai melhorar o desempenho entre cliente e servidor. Essa configuração controla o número padrão de bytes de mensagens de processamento em lote.
Não são feitas tentativas de processar bytes de mensagem maiores que essa contagem de bytes.
As solicitações enviadas aos corretores conterão múltiplos lotes, que conterão uma solicitação para cada partição.
Valores de lote menores são menos usados e podem reduzir o throughput (0 usará apenas processamento em lote). Valores de lote maiores desperdiçam mais espaço de memória, então você precisa alocar memória para valores específicos de lote. | | client.id | String | | Média | Essa string é enviada ao servidor quando uma solicitação é feita. O objetivo é conseguir rastrear a origem das requisições para permitir que alguns aplicativos fora da lista de IP/Porta enviem informações. Este app pode configurar qualquer string porque não tem função além de gravar e gravar | | linger.ms | Longas | 0 | Média | O grupo de produtores agregará quaisquer mensagens que chegarem entre a solicitação e o envio, registrando um lote separado de solicitações. Normalmente, isso só acontece quando o registro é gerado mais rápido que a taxa de envio. No entanto, sob certas condições, o cliente vai querer reduzir o número de solicitações, ou até mesmo uma carga moderada. Essa configuração será feita adicionando um pequeno atraso – ou seja, em vez de enviar um disco imediatamente, o produtor aguardará um determinado tempo de atraso para permitir que outros registros de mensagem sejam enviados, que podem ser agrupados. Isso pode ser considerado um algoritmo semelhante ao TCP Nagle. Essa configuração estabelece um limite de latência maior para o batching: uma vez que obtemos o tamanho batch.de uma partição, ela enviará imediatamente independentemente dessa configuração, mas se recebermos uma mensagem com uma contagem de bytes muito menor que essa configuração, precisamos "demorar" um tempo específico para receber mais mensagens. Essa configuração é 0 por padrão, ou seja, sem atraso. Definir linger.ms=5, por exemplo, reduzirá o número de solicitações, mas ao mesmo tempo aumentará o atraso em 5ms. | | max.request.size | int | 1028576 | Média | O número máximo de bytes solicitados. Essa também é uma cobertura eficaz para o tamanho máximo registrado. Nota: O servidor possui sua própria substituição dos tamanhos dos registros de mensagens, que são diferentes dessa configuração. Essa configuração limita o número de solicitações que os produtores podem enviar em massa ao mesmo tempo para evitar um grande número de solicitações. | | receber.buffer.bytes | int | 32768 | Média | Tamanho do receivecache do TCP, que é usado ao ler dados | | enviar.buffer.bytes | int | 131072 | Média | Tamanho do cache de envio TCP, que é usado ao enviar dados | | timeout.ms | int | 30000 | Média | Essa opção de configuração controla o tempo máximo que o servidor espera pela confirmação dos seguidores. Se o número de solicitações confirmadas não for atendido dentro desse período, um erro é retornado. Esse limite de tempo é medido do lado do servidor e não possui latência de rede, incluindo requisições | | bloquear.on.buffer.full | Booleano | true | baixo | Quando nosso cache de memória acaba, precisamos parar de receber novos registros de mensagens ou lançar erros. Por padrão, isso está configurado como verdadeiro, porém alguns bloqueios podem não valer a pena esperar, então é melhor lançar um erro imediatamente. Isso ocorre quando definido como falso: o produtor lança um erro de exceção: BufferExhaustedException se o registro foi enviado e o cache estiver cheio | | metadata.fetch.timeout.ms | Longas | 60000 | baixo | Refere-se aos dados da primeira vez de alguns elementos que obtivemos. Os elementos incluem: tópico, hospedeiro, partições. Essa configuração se refere ao tempo necessário para o elemento ser concluído com sucesso de acordo com a busca, caso contrário, uma exceção será enviada ao cliente. | | metadata.max.age.ms | Longas | 300000 | baixo | Tempo em microssegundos é o intervalo no qual forçamos a atualização dos metadados. Mesmo que não vejamos mudanças na liderança da partição. | | metric.reporters | Lista | [] | baixo | Uma lista de classes usadas para medir métricas. Implementar a interface MetricReporter permitirá a adição de classes que mudam conforme novas métricas são geradas. O JmxReporter sempre incluirá uma forma de registrar estatísticas JMX | | metrics.num.samples | int | 2 | baixo | O número de amostras usadas para manter métricas | | metrics.sample.window.ms | Longas | 30000 | baixo | O sistema de Métricas mantém um número configurável de amostras em um tamanho de janela corrigível. Essa configuração configura, por exemplo, o tamanho da janela. Podemos manter duas amostras durante 30 segundos. Quando uma janela é aberta, apagamos e reescrevemos a janela mais antiga | | recoonect.backoff.ms | Longas | 10 | baixo | Quando a conexão falha, o tempo de espera para a reconexão acontece. Isso evita reconexões repetidas dos clientes | | retry.backoff.ms | Longas | 100 | baixo | O tempo de espera antes de tentar tentar novamente um pedido de produtos que não foi aprovado. Evite ficar preso em um ciclo morto de envio e falha.
|
|
Publicado em 24/06/2021 11:47:12
|
|
|
Senhorio