Merge branch 'dev' of http://192.168.2.46:8099/jsowell/jsowell-charger-web into dev

jsowell-admin/src/test/java/SpringBootTestController.java

@@ -106,6 +106,8 @@ import java.nio.charset.StandardCharsets;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @ActiveProfiles("dev")
@@ -332,12 +334,12 @@ public class SpringBootTestController {
 
     @Test
     public void createBalancePaymentRequestTest() {
-        String outMemberId = "ACM40782726";
-        String inMemberId = "0";
-        String transAmt = "124.72";
-        String title = "提取余额到自己账户";
-        String desc = "2024年6月14日14点15分，售后需求：客户需要重新添加结算账户，原账户余额放弃提取";
-        String wechatAppId = wechatAppId1;
+        String outMemberId = "ACM42875164"; // 出账memberId
+        String inMemberId = "0"; // 入账memberId
+        String transAmt = "798.20"; // 金额
+        String title = "提取余额到自己账户"; // 标题
+        String desc = "2024年7月31日08点55分，售后需求：客户重新添加结算账户, 原账户余额无法提取, 由现下打款给客户"; // 描述
+        String wechatAppId = wechatAppId1; // 万车充id
         adapayService.createBalancePaymentRequest(outMemberId, inMemberId, transAmt, title, desc, wechatAppId);
     }
 

jsowell-admin/src/test/java/Test.java

@@ -1,85 +0,0 @@
-import com.google.common.collect.Lists;
-
-import java.util.Comparator;
-import java.util.List;
-import java.util.stream.Collectors;
-
-/**
- * @program: com.cjml.service.partitem
- * @description:
- * @author: dc
- * @create: 2023-09-09 13:46
- **/
-public class Test {
-
-    public static void main(String[] args) {
-        List<Category> testList = Lists.newArrayList();
-        testList.add(new Category("100-50-10", 100, 50, 10));
-        testList.add(new Category("110-30-8", 110, 30, 8));
-        testList.add(new Category("0-10-12", 0, 10, 12));
-
-        testList = testList.stream().sorted(
-                        Comparator.comparing(Category::getCategoryGroupSort, Comparator.reverseOrder())
-                                .thenComparing(Category::getCategorySort, Comparator.reverseOrder())
-                                .thenComparing(Category::getPartSort)
-                )
-                .collect(Collectors.toList());
-
-        System.out.println(testList.get(0).getCategoryName());
-        System.out.println(testList.get(1).getCategoryName());
-        System.out.println(testList.get(2).getCategoryName());
-    }
-
-    static class Category {
-
-        private String categoryName;
-
-        private int categoryGroupSort;
-
-        private int categorySort;
-
-        private int partSort;
-
-        public Category() {
-        }
-
-        public Category(String categoryName, int categoryGroupSort, int categorySort, int partSort) {
-            this.categoryName = categoryName;
-            this.categoryGroupSort = categoryGroupSort;
-            this.categorySort = categorySort;
-            this.partSort = partSort;
-        }
-
-        public String getCategoryName() {
-            return categoryName;
-        }
-
-        public void setCategoryName(String categoryName) {
-            this.categoryName = categoryName;
-        }
-
-        public int getCategoryGroupSort() {
-            return categoryGroupSort;
-        }
-
-        public void setCategoryGroupSort(int categoryGroupSort) {
-            this.categoryGroupSort = categoryGroupSort;
-        }
-
-        public int getCategorySort() {
-            return categorySort;
-        }
-
-        public void setCategorySort(int categorySort) {
-            this.categorySort = categorySort;
-        }
-
-        public int getPartSort() {
-            return partSort;
-        }
-
-        public void setPartSort(int partSort) {
-            this.partSort = partSort;
-        }
-    }
-}

jsowell-admin/src/test/java/rpc/ClientChannelInitializer.java

@@ -0,0 +1,19 @@
+package rpc;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+
+public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
+ 
+    @Override
+    protected void initChannel(SocketChannel socketChannel) throws Exception {
+        ChannelPipeline pipeline = socketChannel.pipeline();
+ 
+        pipeline.addLast(new MessageEncode());
+        pipeline.addLast(new MessageDecode());
+ 
+        pipeline.addLast(new RpcResponseHandler());
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/Message.java

@@ -0,0 +1,11 @@
+package rpc;
+
+import lombok.Data;
+
+@Data
+public abstract class Message {
+ 
+    protected Byte messageType;
+ 
+ 
+}

jsowell-admin/src/test/java/rpc/MessageConstant.java

@@ -0,0 +1,22 @@
+package rpc;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class MessageConstant {
+ 
+    public final static Byte rpcRequest = 1;
+    public final static Byte rpcResponse = 2;
+ 
+    public static Map<Byte, Class<? extends Message>> messageTypeMap = new ConcurrentHashMap<>();
+ 
+    static {
+        messageTypeMap.put(rpcRequest, RpcRequest.class);
+        messageTypeMap.put(rpcResponse, RpcResponse.class);
+    }
+ 
+    public static Class<? extends Message> getMessageClass(Byte messageType){
+        return messageTypeMap.get(messageType);
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/MessageDecode.java

@@ -0,0 +1,45 @@
+package rpc;
+
+import com.jsowell.common.util.bean.SerializationUtil;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.ByteToMessageDecoder;
+
+import java.util.List;
+
+public class MessageDecode extends ByteToMessageDecoder {
+ 
+ 
+    @Override
+    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
+ 
+        // 由于数据包的前4个字节用于记录总数据大小，如果数据不够4个字节，不进行读
+        if(byteBuf.readableBytes() < 4) {
+            return;
+        }
+ 
+        // 标记开始读的位置
+        byteBuf.markReaderIndex();
+ 
+        // 前四个字节记录了数据大小
+        int dataSize = byteBuf.readInt();
+ 
+        // 查看剩余可读字节是否足够，如果不是，重置读取位置，等待下一次解析
+        if(byteBuf.readableBytes() < dataSize) {
+            byteBuf.resetReaderIndex();
+            return;
+        }
+ 
+        // 读取消息类型
+        byte messageType = byteBuf.readByte();
+        // 读取数据, 数组大小需要剔除1个字节的消息类型
+        byte[] data = new byte[dataSize -1];
+ 
+        byteBuf.readBytes(data);
+ 
+        Message message = SerializationUtil.deserialize(MessageConstant.getMessageClass(messageType), data);
+ 
+        list.add(message);
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/MessageEncode.java

@@ -0,0 +1,23 @@
+package rpc;
+
+import com.jsowell.common.util.bean.SerializationUtil;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+public class MessageEncode extends MessageToByteEncoder<Message> {
+ 
+    @Override
+    protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {
+        // 将对象进行序列化
+        byte[] data = SerializationUtil.serialize(message);
+ 
+        // 写数据长度，前4个字节用于记录数据总长度（对象 + 类型（1个字节））
+        byteBuf.writeInt(data.length + 1);
+        // 写记录消息类型，用于反序列选择类的类型
+        byteBuf.writeByte(message.getMessageType());
+        // 写对象
+        byteBuf.writeBytes(data);
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/RpcClient.java

@@ -0,0 +1,52 @@
+package rpc;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+public class RpcClient {
+ 
+    public Channel connect(String host, Integer port) {
+        EventLoopGroup worker = new NioEventLoopGroup();
+        Channel channel = null;
+ 
+        try {
+            Bootstrap bootstrap = new Bootstrap();
+ 
+            bootstrap.group(worker)
+                    .channel(NioSocketChannel.class)
+                    .option(ChannelOption.AUTO_READ, true)
+                    .handler(new ClientChannelInitializer());
+ 
+            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
+ 
+            System.out.println("客户端启动");
+ 
+            channel = channelFuture.channel();
+ 
+            // 添加关闭监听器
+            channel.closeFuture().addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture channelFuture) throws Exception {
+                    System.out.println("关闭客户端");
+                    worker.shutdownGracefully();
+                }
+            });
+ 
+        } catch (Exception e) {
+            e.printStackTrace();
+ 
+            if(channel == null || !channel.isActive()) {
+                worker.shutdownGracefully();
+            } else {
+                channel.close();
+            }
+ 
+        }
+ 
+        return channel;
+    }
+ 
+ 
+}

jsowell-admin/src/test/java/rpc/RpcRequest.java

@@ -0,0 +1,21 @@
+package rpc;
+
+import lombok.Data;
+import lombok.ToString;
+
+import java.util.UUID;
+
+@Data
+@ToString
+public class RpcRequest extends Message{
+ 
+    private String id;
+ 
+    private String param;
+ 
+    public RpcRequest() {
+        this.id = UUID.randomUUID().toString();
+        super.messageType = MessageConstant.rpcRequest;
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/RpcRequestHandler.java

@@ -0,0 +1,36 @@
+package rpc;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.DefaultEventLoopGroup;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+public class RpcRequestHandler extends SimpleChannelInboundHandler<RpcRequest> {
+ 
+    private final static EventLoopGroup worker = new DefaultEventLoopGroup(Runtime.getRuntime().availableProcessors() + 1);
+ 
+ 
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, RpcRequest msg) throws Exception {
+ 
+        // 为避免占用网络io，此处异步进行处理
+        worker.submit(() -> {
+            System.out.println("[RpcRequestHandler] "+ Thread.currentThread().getName()  +" 处理请求，msg： " + msg);
+ 
+            // 模拟处理耗时
+            try {
+                Thread.sleep(3000);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+ 
+            RpcResponse rpcResponse = new RpcResponse();
+            rpcResponse.setId(msg.getId());
+            rpcResponse.setResult("处理" + msg.getParam());
+ 
+            ctx.writeAndFlush(rpcResponse);
+        });
+ 
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/RpcResponse.java

@@ -0,0 +1,18 @@
+package rpc;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString
+public class RpcResponse extends Message{
+ 
+    private String id;
+ 
+    private String result;
+ 
+    public RpcResponse() {
+        super.messageType = MessageConstant.rpcResponse;
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/RpcResponseHandler.java

@@ -0,0 +1,23 @@
+package rpc;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+public class RpcResponseHandler extends SimpleChannelInboundHandler<RpcResponse> {
+ 
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception {
+        // 根据请求id，在集合中找到与外部线程通信的SyncPromise对象
+        SyncPromise syncPromise = RpcUtil.getSyncPromiseMap().get(msg.getId());
+ 
+        if(syncPromise != null) {
+            // 设置响应结果
+            syncPromise.setRpcResponse(msg);
+ 
+            // 唤醒外部线程
+            syncPromise.wake();
+        }
+ 
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/RpcServer.java

@@ -0,0 +1,55 @@
+package rpc;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+public class RpcServer {
+ 
+    public void bind(Integer port) {
+ 
+        EventLoopGroup parent = new NioEventLoopGroup();
+        EventLoopGroup child = new NioEventLoopGroup();
+        Channel channel = null;
+ 
+        try{
+            ServerBootstrap serverBootstrap = new ServerBootstrap();
+ 
+            serverBootstrap.group(parent, child)
+                    .channel(NioServerSocketChannel.class)
+                    .option(ChannelOption.SO_BACKLOG, 1024)
+                    .childHandler(new ServerChannelInitializer());
+ 
+            ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
+ 
+            System.out.println("server启动");
+ 
+            // 非阻塞等待关闭
+            channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {
+                @Override
+                public void operationComplete(ChannelFuture channelFuture) throws Exception {
+                    System.out.println("server关闭");
+                    parent.shutdownGracefully();
+                    child.shutdownGracefully();
+                }
+            });
+ 
+            channel = channelFuture.channel();
+ 
+        } catch (Exception e) {
+ 
+            e.printStackTrace();
+ 
+            if(channel == null || !channel.isActive()) {
+                System.out.println("server关闭");
+                parent.shutdownGracefully();
+                child.shutdownGracefully();
+            } else {
+                channel.close();
+            }
+        }
+    }
+ 
+ 
+}

jsowell-admin/src/test/java/rpc/RpcUtil.java

@@ -0,0 +1,63 @@
+package rpc;
+
+import io.netty.channel.Channel;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public class RpcUtil {
+ 
+    private final static Map<String, SyncPromise> syncPromiseMap =  new ConcurrentHashMap<>();
+ 
+    private final static Channel channel;
+ 
+    static{
+        channel = new RpcClient().connect("127.0.0.1", 8888);
+    }
+ 
+    public static RpcResponse send(RpcRequest rpcRequest, long timeout, TimeUnit unit) throws Exception{
+ 
+        if(channel == null) {
+            throw new NullPointerException("channel");
+        }
+ 
+        if(rpcRequest == null) {
+            throw new NullPointerException("rpcRequest");
+        }
+ 
+        if(timeout <= 0) {
+            throw new IllegalArgumentException("timeout must greater than 0");
+        }
+ 
+        // 创造一个容器，用于存放当前线程与rpcClient中的线程交互
+        SyncPromise syncPromise = new SyncPromise();
+        syncPromiseMap.put(rpcRequest.getId(), syncPromise);
+ 
+        // 发送消息，此处如果发送玩消息并且在get之前返回了结果，下一行的get将不会进入阻塞，也可以顺利拿到结果
+        channel.writeAndFlush(rpcRequest);
+ 
+        // 等待获取结果
+        RpcResponse rpcResponse = syncPromise.get(timeout, unit);
+ 
+        if(rpcResponse == null) {
+            if(syncPromise.isTimeout()) {
+                throw new TimeoutException("等待响应结果超时");
+            } else{
+                throw new Exception("其他异常");
+            }
+        }
+ 
+        // 移除容器
+        syncPromiseMap.remove(rpcRequest.getId());
+ 
+        return rpcResponse;
+    }
+ 
+    public static Map<String, SyncPromise> getSyncPromiseMap(){
+        return syncPromiseMap;
+    }
+ 
+ 
+}

jsowell-admin/src/test/java/rpc/ServerChannelInitializer.java

@@ -0,0 +1,19 @@
+package rpc;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.socket.SocketChannel;
+
+public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
+ 
+    @Override
+    protected void initChannel(SocketChannel socketChannel) throws Exception {
+        ChannelPipeline pipeline = socketChannel.pipeline();
+ 
+        pipeline.addLast(new MessageEncode());
+        pipeline.addLast(new MessageDecode());
+ 
+        pipeline.addLast(new RpcRequestHandler());
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/SyncPromise.java

@@ -0,0 +1,49 @@
+package rpc;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class SyncPromise {
+ 
+    // 用于接收结果
+    private RpcResponse rpcResponse;
+ 
+    private final CountDownLatch countDownLatch = new CountDownLatch(1);
+    
+    // 用于判断是否超时
+    private boolean isTimeout = false;
+ 
+    /**
+     * 同步等待返回结果
+     */
+    public RpcResponse get(long timeout, TimeUnit unit) throws InterruptedException {
+        // 等待阻塞，超时时间内countDownLatch减到0，将提前唤醒，以此作为是否超时判断
+        boolean earlyWakeUp = countDownLatch.await(timeout, unit);
+ 
+        if(earlyWakeUp) {
+            // 超时时间内countDownLatch减到0，提前唤醒，说明已有结果
+            return rpcResponse;
+        } else {
+            // 超时时间内countDownLatch没有减到0，自动唤醒，说明超时时间内没有等到结果
+            isTimeout = true;
+            return null;
+        }
+    }
+ 
+    public void wake() {
+        countDownLatch.countDown();
+    }
+ 
+    public RpcResponse getRpcResponse() {
+        return rpcResponse;
+    }
+ 
+    public void setRpcResponse(RpcResponse rpcResponse) {
+        this.rpcResponse = rpcResponse;
+    }
+ 
+    public boolean isTimeout() {
+        return isTimeout;
+    }
+ 
+}

jsowell-admin/src/test/java/rpc/TestRpcClient.java

@@ -0,0 +1,48 @@
+package rpc;
+
+import java.util.concurrent.TimeUnit;
+
+public class TestRpcClient {
+    public static void main(String[] args) throws Exception{
+        //Channel channel = new RpcClient().connect("127.0.0.1", 8888);
+
+        Thread thread1 = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                RpcRequest rpcRequest = new RpcRequest();
+                rpcRequest.setParam("参数1");
+
+                try {
+                    System.out.println("thread1发送请求");
+                    RpcResponse rpcResponse = RpcUtil.send(rpcRequest, 5, TimeUnit.SECONDS);
+                    System.out.println("thread1处理结果：" + rpcResponse);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+
+        Thread thread2 = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                RpcRequest rpcRequest2 = new RpcRequest();
+                rpcRequest2.setParam("参数2");
+
+                try {
+                    System.out.println("thread2发送请求");
+                    RpcResponse rpcResponse = RpcUtil.send(rpcRequest2, 2, TimeUnit.SECONDS);
+                    System.out.println("thread2处理结果：" + rpcResponse);
+                } catch (Exception e) {
+                    throw new RuntimeException(e);
+                }
+            }
+        });
+
+        // 休眠一下，等待客户端与服务端进行连接
+        Thread.sleep(1000);
+
+        thread1.start();
+        thread2.start();
+
+    }
+}

jsowell-admin/src/test/java/rpc/TestRpcServer.java

@@ -0,0 +1,7 @@
+package rpc;
+
+public class TestRpcServer {
+    public static void main(String[] args) {
+        new RpcServer().bind(8888);
+    }
+}