I wrote a timer in kafka's code to send a heartbeat to the producer.
@ThingworxServiceDefinition(name = "ReceiverStringMessage", description = "Receiver a string message to the specified Kafka")
@ThingworxServiceResult(name = "result", description = "", baseType = "STRING")
public String ReceiverStringMessage() throws Exception {
//String results = this.utils.ReceiverStringMessage(entity,strMap);
String msg = "";
//String state = "0";
try {
logger.info("开始接收数据:" + entity.getGetServerAddress() + "=====" + entity.getTopic() + "===" + entity.getZookeeperSyncTimeMs() + "===" + entity.getReceiveTimeout() + "===" + entity.getAutoCommitInterval()+"==="+entity.getFetch());
consumer.subscribe(Collections.singletonList(entity.getTopic()));
logger.info("consumer 订阅成功");
//ConsumerRecords<String, String> records = consumer.poll(entity.getPollTimeout());
logger.info("consumer 提取成功");
//executor = new ScheduledThreadPoolExecutor(1);
//Task5 task5 = new Task5();
//mapMsg = new HashMap<String,String>();
try{
if(!atomic.get()){
timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
//KafkaReceiverThing kafkathing;
@Override
public void run() {
String message = "";
try{
ConsumerRecords<String, String> records = consumer.poll(entity.getPollTimeout());
logger.info("consumer 提取成功");
for (ConsumerRecord<String, String> record : records) {
if(record.value() != null && record.value() != "" ){
message += record.value() + ";";
}
}
if(message != null && message != "" ){
logger.info("数据获取成功111111:" + message);
}
json = new JSONObject();
if(message != null && message != "" ){
json.put("state", state);
json.put("msg", message);
logger.info("收到的消息:"+json.toString());
strMap.put("mapmsg2", json.toString());
logger.info("map value:"+strMap.get("mapmsg2"));
results = strMap.get("mapmsg2");
logger.info("收到的数据有值:"+results);
//logger.info("property value:"+GetPropertyDefinition("strvalue"));
//logger.info("property value:"+GetPropertyDefinition("strvalue").getFirstRow().getStringValue("value"));
//String msg = getPropertyValue("strvalue").toString();
//logger.info("property value:"+msg);
setPropertyValue("strvalue", new StringPrimitive(results));
//logger.info("property length:"+GetPropertyDefinition("strvalue").getLength());
}else{
json.put("state", state);
json.put("msg", message);
strMap.put("mapmsg2", json.toString());
results = strMap.get("mapmsg2");
logger.info("收到的数据为空:"+results);
}
} catch (JSONException e) {
e.printStackTrace();
} catch (Exception e) {
// 添加kafkathing.GetPropertyDefinition("strvalue").getReturnValue()方法后抛出的异常
e.printStackTrace();
}
}
}, 0, 10000);
}
} catch (WakeupException e){
if (!atomic.get()) throw e;
}
} catch (Exception e) {
state = "1";
msg = "获取数据失败";
logger.info("获取数据失败222222：" + e.getLocalizedMessage());
}
logger.info("接收到的数据为："+strMap.get("mapmsg2"));
return results;
}
