我已经实现了一个java消费者,它使用来自kafka主题的消息,然后将这些消息与post请求一起发送到restapi。
while (true) {
ConsumerRecords<String, Object> records = consumer.poll(200);
for (ConsumerRecord<String, Object> record : records) {
CloseableHttpClient httpClient = HttpClientBuilder.create().build();
Object message = record.value();
JSONObject jsonObj = new JSONObject(message.toString());
try {
HttpPost request = new HttpPost(this.getConsumerProperties().getProperty("api.url"));
StringEntity params = new StringEntity(message.toString());
request.addHeader("content-type", "application/json");
request.addHeader("Accept", "application/json");
request.setEntity(params);
CloseableHttpResponse response = httpClient.execute(request);
HttpEntity entity = response.getEntity();
String responseString = EntityUtils.toString(entity, "UTF-8");
System.out.println(message.toString());
System.out.println(responseString);
} catch(Exception ex) {
ex.printStackTrace();
} finally {
try {
httpClient.close();
} catch(IOException ex) {
ex.printStackTrace();
}
}
}
}
假设消息已被使用,但java类未能访问restapi。消息将永远不会被传递,但它将被标记为已使用。处理此类案件的最佳方法是什么?当且仅当restapi的响应成功时,我能以某种方式确认消息吗?
1条答案
按热度按时间fykwrbwg1#
在使用者属性中,将enable.auto.commit设置为false。这意味着消费者有责任承担补偿。
因此,在上面的示例中,基于response.statuscode,您可以通过调用consumer.commitasync()来选择提交偏移量。