消息推送系统

@Configuration

public class KafkaConfig {

@Bean

public ProducerFactory producerFactory() {

Map configProps = new HashMap<>();

configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

return new DefaultKafkaProducerFactory<>(configProps);

}

@Bean

public KafkaTemplate kafkaTemplate() {

return new KafkaTemplate<>(producerFactory());

}

}

]]>

@Service

public class MessageConsumer implements Runnable {

private final KafkaListenerContainerFactory kafkaListenerContainerFactory;

public MessageConsumer(KafkaListenerContainerFactory kafkaListenerContainerFactory) {

this.kafkaListenerContainerFactory = kafkaListenerContainerFactory;

}

@Override

智慧学工系统

public void run() {

ConcurrentMessageListenerContainer container =

(ConcurrentMessageListenerContainer) kafkaListenerContainerFactory.createListenerContainer();

container.setupMessageListener(new MessageListener() {

@Override

public void onMessage(ConsumerRecord record) {

System.out.println("Received message: " + record.value());

}

});

container.start();

}

}

]]>