目前项目预设后端用的spring,因为要调用算法所以也引用了flask框架,spring和flask之间交互用的Kafka消息队列,前端开发人员问我“有哪些数据用Kafka发送,数据格式都是什么?”
在spring后端这边就是获取前端用户输入的各种算法参数(string类型的),然后把参数借助Kafka传递给flask,flask调用模型算法计算完,再把结果值传递回spring,这个数据格式依然是string吗?或者会被转成json? (我瞎说的)我也不懂这些,对Kafka完全不了解
目前项目预设后端用的spring,因为要调用算法所以也引用了flask框架,spring和flask之间交互用的Kafka消息队列,前端开发人员问我“有哪些数据用Kafka发送,数据格式都是什么?”
在spring后端这边就是获取前端用户输入的各种算法参数(string类型的),然后把参数借助Kafka传递给flask,flask调用模型算法计算完,再把结果值传递回spring,这个数据格式依然是string吗?或者会被转成json? (我瞎说的)我也不懂这些,对Kafka完全不了解
回答:
在使用Kafka队列交互数据时,我们需要确保数据格式的正确性和易于解析性。一种常见的做法是将数据格式化成JSON格式,并在Spring和Flask之间进行传递。以下是实现步骤:
@Service
public class KafkaProducerService {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void sendMessage(String topic, String message) {
kafkaTemplate.send(topic, message);
}
}
from kafka import KafkaConsumer
consumer = KafkaConsumer('topic_name',
bootstrap_servers=['localhost:9092'],
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
for message in consumer:
print(message.value)
# 在这里进行数据处理
在Spring中:
ObjectMapper objectMapper = new ObjectMapper();
String jsonString = objectMapper.writeValueAsString(algorithmParams);
kafkaProducerService.sendMessage("topic_name", jsonString);
在Flask中:
import json
message_object = json.loads(message.value)
总结:
通过Kafka队列让Spring和Flask交互需要进行以下步骤: