消息推送系统

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

public class KafkaProducerExample {

public static void main(String[] args) {

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

Producer producer = new KafkaProducer<>(props);

ProducerRecord record = new ProducerRecord<>("hospital-auction-topic", "{\"auctionId\": \"123456\", \"title\": \"医疗设备采购招标\", \"status\": \"published\"}");

producer.send(record);

producer.close();

}

}

import org.apache.kafka.clients.consumer.*;

import java.util.Properties;

import java.util.Arrays;

public class KafkaConsumerExample {

public static void main(String[] args) {

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("group.id", "hospital-auction-group");

props.put("enable.auto.commit", "true");

props.put("auto.offset.reset", "earliest");

props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Consumer consumer = new KafkaConsumer<>(props);

consumer.subscribe(Arrays.asList("hospital-auction-topic"));

while (true) {

ConsumerRecords records = consumer.poll(100);

for (ConsumerRecord record : records) {

System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

// 这里可以添加业务逻辑，比如更新数据库或通知相关人员

}

}

}

}