使用R2DBC访问MySQL测试

pom.xml

@@ -30,23 +30,49 @@
     </dependencyManagement>
 
     <dependencies>
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-starter</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-webflux</artifactId>
-        </dependency>
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-test</artifactId>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-webflux</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-starter</artifactId>
+		</dependency>
+		<dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
         </dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-data-r2dbc</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>dev.miku</groupId>
+			<artifactId>r2dbc-mysql</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>mysql</groupId>
+			<artifactId>mysql-connector-java</artifactId>
+			<scope>runtime</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.projectlombok</groupId>
+			<artifactId>lombok</artifactId>
+			<optional>true</optional>
+		</dependency>
+
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>io.projectreactor</groupId>
+			<artifactId>reactor-test</artifactId>
+			<scope>test</scope>
+		</dependency>
     </dependencies>
 
     <build>

src/main/java/com/caimei365/user/controller/TestApi.java

@@ -1,8 +1,10 @@
 package com.caimei365.user.controller;
 
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import com.caimei365.user.entity.BaseUser;
+import com.caimei365.user.service.BaseUserService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+import reactor.core.publisher.Mono;
 
 /**
  * Description
@@ -10,14 +12,36 @@ import org.springframework.web.bind.annotation.RestController;
  * @author : Charles
  * @date : 2021/02/20
  */
-@RequestMapping("/user")
+
 @RestController
+@RequiredArgsConstructor
+@RequestMapping("/user")
 public class TestApi {
 
+    private final BaseUserService baseUserService;
+
     @GetMapping("/test")
     public String getTestString() {
         return "test user";
     }
 
+    @GetMapping("/get")
+    public Mono<BaseUser> findById(Integer id) {
+        return baseUserService.findById(id);
+    }
+
+    @PostMapping("/update")
+    public Mono<Integer> updateById(BaseUser user) {
+        return baseUserService.updateById(user);
+    }
 
+    @PostMapping("/insert")
+    public Mono<Void> insertUser(BaseUser user) {
+        return baseUserService.insertUser(user);
+    }
+
+    @PostMapping("/delete")
+    public Mono<Void> deleteById(Integer id) {
+        return baseUserService.deleteById(id);
+    }
 }

src/main/java/com/caimei365/user/dao/BaseUserDao.java

@@ -0,0 +1,53 @@
+package com.caimei365.user.dao;
+
+import com.caimei365.user.entity.BaseUser;
+import org.springframework.data.r2dbc.repository.Modifying;
+import org.springframework.data.r2dbc.repository.Query;
+import org.springframework.data.repository.reactive.ReactiveCrudRepository;
+import org.springframework.stereotype.Repository;
+import reactor.core.publisher.Mono;
+
+/**
+ * Description
+ *
+ * @author : Charles
+ * @date : 2021/3/2
+ */
+@Repository
+public interface BaseUserDao extends ReactiveCrudRepository<BaseUser, Integer> {
+    /**
+     * 根据用户id查询用户
+     * @param id userId
+     * @return BaseUser
+     */
+    @Override
+    @Query("select userId as id, userMobile as mobile, userName as name from base_user where userId= :id")
+    Mono<BaseUser> findById(Integer id);
+
+    /**
+     * 更新用户名
+     * @param id userId
+     * @param name userName
+     */
+    @Modifying
+    @Query("update base_user set userName= :name where userId= :id")
+    Mono<Integer> updateNameById(Integer id, String name);
+
+    /**
+     * 新增用户
+     * @param name userName
+     * @param mobile userMobile
+     */
+    @Modifying
+    @Query("insert into base_user(userName, userMobile) values (:name, :mobile)")
+    Mono<Void> insertUser(String name, String mobile);
+
+    /**
+     * 根据用户id删除用户
+     * @param id userId
+     */
+    @Override
+    @Modifying
+    @Query("delete from base_user where userId= :id")
+    Mono<Void> deleteById(Integer id);
+}

src/main/java/com/caimei365/user/entity/BaseUser.java

@@ -0,0 +1,16 @@
+package com.caimei365.user.entity;
+
+import lombok.Data;
+
+/**
+ * Description
+ *
+ * @author : Charles
+ * @date : 2021/3/2
+ */
+@Data
+public class BaseUser {
+    private Integer id;
+    private String name;
+    private String mobile;
+}

src/main/java/com/caimei365/user/service/BaseUserService.java

@@ -0,0 +1,39 @@
+package com.caimei365.user.service;
+
+import com.caimei365.user.dao.BaseUserDao;
+import com.caimei365.user.entity.BaseUser;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import reactor.core.publisher.Mono;
+
+/**
+ * Description
+ *
+ * @author : Charles
+ * @date : 2021/3/2
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class BaseUserService {
+
+    private final BaseUserDao baseUserDao;
+
+    public Mono<BaseUser> findById (Integer id) {
+        return baseUserDao.findById(id);
+    }
+
+    public Mono<Integer> updateById(BaseUser user){
+        return baseUserDao.updateNameById(user.getId(), user.getName());
+    }
+
+    public Mono<Void> insertUser(BaseUser user){
+        return baseUserDao.insertUser(user.getName(), user.getMobile());
+    }
+
+    public Mono<Void> deleteById(Integer id) {
+        return baseUserDao.deleteById(id);
+    }
+
+}

src/main/resources/application.yml

@@ -5,6 +5,10 @@ server:
 spring:
   application:
     name: @artifactId@
+  r2dbc:
+    url: r2dbcs:mysql://192.168.2.100:3306/caimei?characterEncoding=UTF8&serverTimezone=Asia/Shanghai
+    username: developer
+    password: 05bZ/OxTB:X+yd%1
 
 # 指定服务注册中心的地址
 eureka:
@@ -14,3 +18,10 @@ eureka:
   client:
     service-url:                  # 设置服务注册中心地址
       defaultZone: http://localhost:18000/eureka/
+
+
+#日志配置
+logging:
+  level:
+    root: info
+    dev.miku.r2dbc.mysql.client.ReactorNettyClient: debug

src/test/java/com/caimei365/user/TestReactive.java

@@ -0,0 +1,93 @@
+package com.caimei365.user;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Description
+ *
+ * @author : Charles
+ * @date : 2021/3/1
+ */
+@SpringBootTest
+class TestReactive {
+    @Test
+    void test1() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .subscribe(System.out::println);
+    }
+
+    @Test
+    void test2() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .subscribe(System.out::println,
+                error -> System.err.println("Error: " + error));
+    }
+
+    @Test
+    void test3() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .onErrorReturn("Divided by zero :(")
+            .subscribe(System.out::println);
+    }
+
+    @Test
+    void test4() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .onErrorResume(e -> System.out::println)
+            .subscribe(System.out::println);
+    }
+    @Test
+    void test5() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .doOnError(error -> System.out.println("we got the error: "+ error))
+            .subscribe(System.out::println);
+    }
+
+    @Test
+    void test6() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .doFinally(error -> System.out.println("Finally，I will make sure to do something:"+error))
+            .subscribe(System.out::println);
+    }
+
+    @Test
+    void test7() {
+        AtomicBoolean isDisposed = new AtomicBoolean();
+        Disposable disposableInstance = new Disposable() {
+            @Override
+            public void dispose() {
+                isDisposed.set(true);
+            }
+            @Override
+            public String toString() {
+                return "DISPOSABLE";
+            }
+        };
+        Flux.using(
+            () -> disposableInstance,
+            disposable -> Flux.just(disposable.toString()),
+            Disposable::dispose)
+            .subscribe(System.out::println);
+    }
+
+    @Test
+    void test8() {
+        Flux.just(10, 5, 0)
+            .map(i -> "100 / " + i + " = " + (10 / i))
+            .retry(1)
+            .elapsed()
+            .subscribe(System.out::println);
+    }
+
+}