事件驱动架构实践

步骤一：准备工作

1. 安装Kafka：确保你已经安装并运行了Kafka。如果没有，可以参考Kafka官网的安装指南。
2. 创建Kafka Topic：创建一个名为“events”的topic。

sh
复制代码
kafka-topics.sh --create --topic events --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

步骤二：创建Spring Boot项目

步骤三：配置Kafka

yaml
复制代码
spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

步骤四：创建Producer和Consumer

Producer

java
复制代码
package com.example.kafkaeventdriven;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Service
public class KafkaProducerService {
    private static final String TOPIC = "events";
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void sendMessage(String message) {
        kafkaTemplate.send(TOPIC, message);
    }
}

Consumer

java
复制代码
package com.example.kafkaeventdriven;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
@Service
public class KafkaConsumerService {
    @KafkaListener(topics = "events", groupId = "my-group")
    public void listen(String message) {
        System.out.println("Received Message: " + message);
        // 在这里处理消息
    }
}

步骤五：创建控制器

java
复制代码
package com.example.kafkaeventdriven;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
@RestController
public class EventController {
    @Autowired
    private KafkaProducerService producerService;
    @GetMapping("/publish")
    public String publishMessage(@RequestParam("message") String message) {
        producerService.sendMessage(message);
        return "Message published successfully";
    }
}

步骤六：运行程序

总结