Este conteúdo foi traduzido por máquina para sua conveniência e a Huawei Cloud não pode garantir que o conteúdo foi traduzido com precisão. Para exibir o conteúdo original, use o link no canto superior direito para mudar para a página em inglês.
Central de ajuda/
Distributed Message Service for Kafka/
Melhores práticas/
Otimização da sondagem de mensagens dos consumidores de DMS for Kafka
Atualizado em 2024-09-09 GMT+08:00
Otimização da sondagem de mensagens dos consumidores de DMS for Kafka
Visão geral
No SDK do Kafka nativo fornecido pelo DMS for Kafka, os consumidores podem personalizar a duração da extração de mensagens. Para extrair mensagens por um longo período, os consumidores só precisam definir o parâmetro do método poll(long) para um valor adequado. No entanto, essas conexões persistentes podem causar pressão sobre o cliente e o servidor, especialmente quando o número de partições é grande e vários threads são ativados para cada consumidor.
Conforme mostrado em Figura 1, o tópico contém várias partições e vários consumidores no grupo de consumidores consomem os recursos ao mesmo tempo. Cada thread está em uma conexão persistente. Quando há poucas ou nenhuma mensagem no tópico, a conexão persiste e todos os consumidores extraem mensagens continuamente, o que causa um desperdício de recursos.
Solução de otimização
Quando vários threads são ativados para acesso simultâneo, se não houver nenhuma mensagem no tópico, apenas um thread será necessário para pesquisar mensagens em cada partição. Quando uma mensagem é encontrada pelo thread de sondagem, outros threads podem ser ativados para consumir a mensagem para respostas rápidas, como mostrado em Figura 2.
Essa solução é aplicável a cenários com baixos requisitos de consumo de mensagens em tempo real. Se for necessário o consumo de mensagens em tempo quase real, recomenda-se que todos os consumidores estejam no estado ativo.
O número de consumidores e o número de partições não são necessariamente os mesmos. O método poll (long) do Kafka ajuda a implementar funções como aquisição de mensagens, balanceamento de partições e detecção de pulsação entre os consumidores e os agentes do Kafka.
Portanto, em cenários em que os requisitos de consumo de mensagens em tempo real são baixos e há um pequeno número de mensagens, alguns consumidores podem estar no estado de espera.
Código de exemplo
A seguir descreve apenas o código relacionado à ativação e suspensão do thread do consumidor. Para executar toda a demonstração, faça o download do pacote de código de exemplo completo e consulte o Guia de desenvolvedor para implementar e executar o código.
Exemplo de código para consumir mensagens:
package com.huawei.dms.kafka;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.Properties;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.log4j.Logger;
public class DmsKafkaConsumeDemo
{
private static Logger logger = Logger.getLogger(DmsKafkaProduceDemo.class);
public static void WorkerFunc(int workerId, KafkaConsumer<String, String> kafkaConsumer) throws IOException
{
Properties consumerConfig = Config.getConsumerConfig();
RecordReceiver receiver = new RecordReceiver(workerId, kafkaConsumer, consumerConfig.getProperty("topic"));
while (true)
{
ConsumerRecords<String, String> records = receiver.receiveMessage();
Iterator<ConsumerRecord<String, String>> iter = records.iterator();
while (iter.hasNext())
{
ConsumerRecord<String, String> cr = iter.next();
System.out.println("Thread" + workerId + " recievedrecords" + cr.value());
logger.info("Thread" + workerId + " recievedrecords" + cr.value());
}
}
}
public static KafkaConsumer<String, String> getConsumer() throws IOException
{
Properties consumerConfig = Config.getConsumerConfig();
consumerConfig.put("ssl.truststore.location", Config.getTrustStorePath());
System.setProperty("java.security.auth.login.config", Config.getSaslConfig());
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(consumerConfig);
kafkaConsumer.subscribe(Arrays.asList(consumerConfig.getProperty("topic")),
new ConsumerRebalanceListener()
{
@Override
public void onPartitionsRevoked(Collection<TopicPartition> arg0)
{
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> tps)
{
}
});
return kafkaConsumer;
}
public static void main(String[] args) throws IOException
{
//Create a consumer for the current consumer group.
final KafkaConsumer<String, String> consumer1 = getConsumer();
Thread thread1 = new Thread(new Runnable()
{
public void run()
{
try
{
WorkerFunc(1, consumer1);
}
catch (IOException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
final KafkaConsumer<String, String> consumer2 = getConsumer();
Thread thread2 = new Thread(new Runnable()
{
public void run()
{
try
{
WorkerFunc(2, consumer2);
}
catch (IOException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
final KafkaConsumer<String, String> consumer3 = getConsumer();
Thread thread3 = new Thread(new Runnable()
{
public void run()
{
try
{
WorkerFunc(3, consumer3);
}
catch (IOException e)
{
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
// //Start threads.
thread1.start();
thread2.start();
thread3.start();
try
{
Thread.sleep(5000);
}
catch (InterruptedException e)
{
e.printStackTrace();
}
//Add threads.
try
{
thread1.join();
thread2.join();
thread3.join();
}
catch (InterruptedException e)
{
e.printStackTrace();
}
}
}
Exemplo de código para gerenciamento de threads do consumidor:
O código de exemplo fornece apenas ideias de design simples. Os desenvolvedores podem otimizar os mecanismos de ativação e suspensão do thread com base em cenários reais.
package com.huawei.dms.kafka;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;
public class RecordReceiver
{
private static Logger logger = Logger.getLogger(DmsKafkaProduceDemo.class);
//Interval time of polling
public static final int WAIT_SECONDS = 10 * 1000;
protected static final Map<String, Object> sLockObjMap = new HashMap<String, Object>();
protected static Map<String, Boolean> sPollingMap = new ConcurrentHashMap<String, Boolean>();
protected Object lockObj;
protected String topicName;
protected KafkaConsumer<String, String> kafkaConsumer;
protected int workerId;
public RecordReceiver(int id, KafkaConsumer<String, String> kafkaConsumer, String queue)
{
this.kafkaConsumer = kafkaConsumer;
this.topicName = queue;
this.workerId = id;
synchronized (sLockObjMap)
{
lockObj = sLockObjMap.get(topicName);
if (lockObj == null)
{
lockObj = new Object();
sLockObjMap.put(topicName, lockObj);
}
}
}
public boolean setPolling()
{
synchronized (lockObj)
{
Boolean ret = sPollingMap.get(topicName);
if (ret == null || !ret)
{
sPollingMap.put(topicName, true);
return true;
}
return false;
}
}
//Wake up all threads.
public void clearPolling()
{
synchronized (lockObj)
{
sPollingMap.put(topicName, false);
lockObj.notifyAll();
System.out.println("Everyone WakeUp and Work!");
logger.info("Everyone WakeUp and Work!");
}
}
public ConsumerRecords<String, String> receiveMessage()
{
boolean polling = false;
while (true)
{
//Check the poll status of threads and hibernate the threads when necessary.
synchronized (lockObj)
{
Boolean p = sPollingMap.get(topicName);
if (p != null && p)
{
try
{
System.out.println("Thread" + workerId + " Have a nice sleep!");
logger.info("Thread" + workerId +" Have a nice sleep!");
polling = false;
lockObj.wait();
}
catch (InterruptedException e)
{
System.out.println("MessageReceiver Interrupted! topicName is " + topicName);
logger.error("MessageReceiver Interrupted! topicName is "+topicName);
return null;
}
}
}
//Start to consume and wake up other threads when necessary.
try
{
ConsumerRecords<String, String> Records = null;
if (!polling)
{
Records = kafkaConsumer.poll(100);
if (Records.count() == 0)
{
polling = true;
continue;
}
}
else
{
if (setPolling())
{
System.out.println("Thread" + workerId + " Polling!");
logger.info("Thread " + workerId + " Polling!");
}
else
{
continue;
}
do
{
System.out.println("Thread" + workerId + " KEEP Poll records!");
logger.info("Thread" + workerId + " KEEP Poll records!");
try
{
Records = kafkaConsumer.poll(WAIT_SECONDS);
}
catch (Exception e)
{
System.out.println("Exception Happened when polling records: " + e);
logger.error("Exception Happened when polling records: " + e);
}
} while (Records.count()==0);
clearPolling();
}
//Acknowledge message consumption.
kafkaConsumer.commitSync();
return Records;
}
catch (Exception e)
{
System.out.println("Exception Happened when poll records: " + e);
logger.error("Exception Happened when poll records: " + e);
}
}
}
}
topicName é o nome do tópico.
Resultados de execução do exemplo de código
[2018-01-25 22:40:51,841] INFO Thread 2 Polling! (com.huawei.dms.kafka.DmsKafkaProduceDemo:119) [2018-01-25 22:40:51,841] INFO Thread2 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:40:52,122] INFO Everyone WakeUp and Work! (com.huawei.dms.kafka.DmsKafkaProduceDemo:69) [2018-01-25 22:40:52,169] INFO Thread2 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:40:52,169] INFO Thread2 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:40:52,216] INFO Thread2 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:40:52,325] INFO Thread 2 Polling! (com.huawei.dms.kafka.DmsKafkaProduceDemo:119) [2018-01-25 22:40:52,325] INFO Thread2 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:40:54,947] INFO Thread1 Have a nice sleep! (com.huawei.dms.kafka.DmsKafkaProduceDemo:87) [2018-01-25 22:40:54,979] INFO Thread3 Have a nice sleep! (com.huawei.dms.kafka.DmsKafkaProduceDemo:87) [2018-01-25 22:41:32,347] INFO Thread2 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:41:42,353] INFO Thread2 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:41:47,816] INFO Everyone WakeUp and Work! (com.huawei.dms.kafka.DmsKafkaProduceDemo:69) [2018-01-25 22:41:47,847] INFO Thread2 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:47,925] INFO Thread 3 Polling! (com.huawei.dms.kafka.DmsKafkaProduceDemo:119) [2018-01-25 22:41:47,925] INFO Thread1 Have a nice sleep! (com.huawei.dms.kafka.DmsKafkaProduceDemo:87) [2018-01-25 22:41:47,925] INFO Thread3 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:41:47,957] INFO Thread2 Have a nice sleep! (com.huawei.dms.kafka.DmsKafkaProduceDemo:87) [2018-01-25 22:41:48,472] INFO Everyone WakeUp and Work! (com.huawei.dms.kafka.DmsKafkaProduceDemo:69) [2018-01-25 22:41:48,503] INFO Thread3 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:48,518] INFO Thread1 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:48,550] INFO Thread2 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:48,597] INFO Thread1 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:48,659] INFO Thread 2 Polling! (com.huawei.dms.kafka.DmsKafkaProduceDemo:119) [2018-01-25 22:41:48,659] INFO Thread2 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128) [2018-01-25 22:41:48,675] INFO Thread3 recievedrecordshello, dms kafka. (com.huawei.dms.kafka.DmsKafkaProduceDemo:32) [2018-01-25 22:41:48,675] INFO Everyone WakeUp and Work! (com.huawei.dms.kafka.DmsKafkaProduceDemo:69) [2018-01-25 22:41:48,706] INFO Thread 1 Polling! (com.huawei.dms.kafka.DmsKafkaProduceDemo:119) [2018-01-25 22:41:48,706] INFO Thread1 KEEP Poll records! (com.huawei.dms.kafka.DmsKafkaProduceDemo:128)
Feedback
Esta página foi útil?
Deixar um comentário
Obrigado por seus comentários. Estamos trabalhando para melhorar a documentação.
O sistema está ocupado. Tente novamente mais tarde.
