data 添加 mysql r2dbc demo

data/data-service/pom.xml

@@ -37,6 +37,21 @@
             <artifactId>spring-boot-starter-test</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>mysql</groupId>
+            <artifactId>mysql-connector-java</artifactId>
+            <scope>runtime</scope>
+        </dependency>
+        <dependency>
+            <groupId>dev.miku</groupId>
+            <artifactId>r2dbc-mysql</artifactId>
+            <version>0.8.2.RELEASE</version>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-amqp</artifactId>

data/data-service/src/main/java/cn/reghao/tnb/data/app/handler/UserProfileHandler.java

@@ -0,0 +1,50 @@
+package cn.reghao.tnb.data.app.handler;
+
+import cn.reghao.tnb.data.app.model.po.UserProfile;
+import cn.reghao.tnb.data.app.repository.UserProfileRepository;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.MediaType;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.server.ServerRequest;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author reghao
+ * @date 2025-05-29 11:37:29
+ */
+@Component
+public class UserProfileHandler {
+    @Autowired
+    private UserProfileRepository userRepository;
+
+    public Mono<ServerResponse> addUser(ServerRequest request) {
+        return ServerResponse.ok()
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(userRepository.saveAll(request.bodyToMono(UserProfile.class)), UserProfile.class);
+    }
+
+    public Mono<ServerResponse> delUser(ServerRequest request) {
+        return userRepository.findById(Integer.parseInt(request.pathVariable("id")))
+                .flatMap(user -> userRepository.delete(user).then(ServerResponse.ok().build()))
+                .switchIfEmpty(ServerResponse.notFound().build());
+    }
+
+    public Mono<ServerResponse> updateUser(ServerRequest request) {
+        return ServerResponse.ok()
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(userRepository.saveAll(request.bodyToMono(UserProfile.class)), UserProfile.class);
+    }
+
+    public Mono<ServerResponse> getAllUser(ServerRequest request) {
+        return ServerResponse.ok()
+                .contentType(MediaType.APPLICATION_JSON)
+                .body(userRepository.findAll(), UserProfile.class);
+    }
+
+    public Mono<ServerResponse> getAllUserStream(ServerRequest request) {
+        return ServerResponse.ok()
+                .contentType(MediaType.TEXT_EVENT_STREAM)
+                .body(userRepository.findAll(), UserProfile.class);
+    }
+}

data/data-service/src/main/java/cn/reghao/tnb/data/app/model/po/UserProfile.java

@@ -0,0 +1,25 @@
+package cn.reghao.tnb.data.app.model.po;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import org.springframework.data.annotation.Id;
+import org.springframework.data.domain.Persistable;
+import org.springframework.data.relational.core.mapping.Table;
+
+/**
+ * @author reghao
+ * @date 2025-05-29 11:08:42
+ */
+@AllArgsConstructor
+@NoArgsConstructor
+@Setter
+@Getter
+@Table("aaa_user_profile")
+public class UserProfile {
+    @Id
+    private Integer id;
+    private String username;
+    private Integer gender;
+}

data/data-service/src/main/java/cn/reghao/tnb/data/app/repository/UserProfileRepository.java

@@ -0,0 +1,11 @@
+package cn.reghao.tnb.data.app.repository;
+
+import cn.reghao.tnb.data.app.model.po.UserProfile;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+
+/**
+ * @author reghao
+ * @date 2025-05-29 11:11:36
+ */
+public interface UserProfileRepository extends ReactiveCrudRepository<UserProfile, Integer> {
+}

data/data-service/src/main/java/cn/reghao/tnb/data/app/config/RouterConfig.java → data/data-service/src/main/java/cn/reghao/tnb/data/app/route/RouterConfig.java

@@ -1,6 +1,7 @@
-package cn.reghao.tnb.data.app.config;
+package cn.reghao.tnb.data.app.route;
 
 import cn.reghao.tnb.data.app.handler.TimeHandler;
+import cn.reghao.tnb.data.app.handler.UserProfileHandler;
 import cn.reghao.tnb.data.app.handler.VideoHandler;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -29,4 +30,16 @@ public class RouterConfig {
                 .andRoute(RequestPredicates.GET("/api/data/video/banner"), videoHandler::hotVideo1)
                 .andRoute(RequestPredicates.GET("/api/data/video/hot1"), videoHandler::hotVideo);
     }
+
+    @Bean
+    RouterFunction<ServerResponse> userRoute(UserProfileHandler userHandler) {
+        return RouterFunctions.nest(
+                RequestPredicates.path("/user"),
+                RouterFunctions.route(RequestPredicates.POST(""), userHandler::addUser)
+                        .andRoute(RequestPredicates.DELETE("/{id}"), userHandler::delUser)
+                        .andRoute(RequestPredicates.PUT(""), userHandler::updateUser)
+                        .andRoute(RequestPredicates.GET(""), userHandler::getAllUser)
+                        .andRoute(RequestPredicates.GET("/stream"), userHandler::getAllUserStream)
+        );
+    }
 }

data/data-service/src/main/java/cn/reghao/tnb/data/app/service/UserProfileService.java

@@ -0,0 +1,43 @@
+package cn.reghao.tnb.data.app.service;
+
+import cn.reghao.tnb.data.app.model.po.UserProfile;
+import cn.reghao.tnb.data.app.repository.UserProfileRepository;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author reghao
+ * @date 2025-05-29 11:31:18
+ */
+@Service
+public class UserProfileService {
+    private final UserProfileRepository userProfileRepository;
+
+    public UserProfileService(UserProfileRepository userProfileRepository) {
+        this.userProfileRepository = userProfileRepository;
+    }
+
+    public Mono<UserProfile> add(UserProfile userProfile) {
+        return userProfileRepository.save(userProfile);
+    }
+
+    public Mono<ResponseEntity<UserProfile>> update(UserProfile userProfile) {
+        return userProfileRepository.findById(userProfile.getId())
+                .flatMap(userProfile1 -> userProfileRepository.save(userProfile))
+                .map(userProfile1 -> new ResponseEntity<>(userProfile1, HttpStatus.OK))
+                .defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
+    }
+
+    public Mono<ResponseEntity<Void>> delete(int id) {
+        return userProfileRepository.findById(id)
+                .flatMap(userProfile -> userProfileRepository.delete(userProfile).then(Mono.just(new ResponseEntity<Void>(HttpStatus.OK))))
+                .defaultIfEmpty(new ResponseEntity<>(HttpStatus.NOT_FOUND));
+    }
+
+    public Flux<UserProfile> find() {
+        return userProfileRepository.findAll();
+    }
+}

data/data-service/src/main/resources/application-dev.yml

@@ -2,6 +2,11 @@ spring:
   cloud:
     discovery:
       enabled: true
+  r2dbc:
+    driver-class-name: com.mysql.jdbc.Driver
+    url: r2dbcs:mysql://127.0.0.1/tnb_content_rdb?SSL=false&sslMode=DISABLED&useUnicode=true&characterEncoding=UTF8&autoReconnect=true
+    username: dev
+    password: Dev@123456
   redis:
     database: 0
     host: 127.0.0.1