下单 打票重试机制 4秒重试 5次 5次后丢失

下单 order数据 不发给生成订单的用户

.gitignore

@@ -8,4 +8,5 @@
 /.gradle/
 /application.pid
 /target/*
-**.jar
+**.jar
+**.log

src/main/java/com/chaozhanggui/system/cashierservice/netty/PushToAppChannelHandlerAdapter.java

@@ -197,6 +197,34 @@ public class PushToAppChannelHandlerAdapter extends NettyChannelHandlerAdapter {
         sendMessage(ctx, str);
     }
 
+
+    /**
+     * 发送给当前用户以外的用户
+     */
+    @Async
+    public void AppSendInfo(String message, String tableId, String userId) {
+        log.info("netty连接 发送消息 除当前用户 tableId:{}  userId:{}", tableId, userId);
+        if (webSocketMap.containsKey(tableId)) {
+            ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(tableId);
+            if (!webSockets.isEmpty()) {
+                for (String user : webSockets.keySet()) {
+                    if (StringUtils.isNotBlank(userId) && userId.equals(user)) {
+                        continue;
+                    }
+                    ChannelHandlerContext ctx = webSockets.get(user);
+                    if (ctx != null) {
+                        sendMesToApp(message, ctx);
+                    } else {
+                        log.info("netty连接 发送消息 除当前用户  userId:{} 失败", user);
+                    }
+                }
+            } else {
+                log.info("netty连接 发送消息 除当前用户 tableId:{} 失败", tableId);
+            }
+        }
+
+    }
+
     @Async
     public void AppSendInfo(String message, String tableId,String userId, boolean userFlag) {
         log.info("netty连接 发送消息 tableId:{}  userId:{} userFlag:{} message:{}",tableId,userId,userFlag, JSONUtil.toJSONString(message));

src/main/java/com/chaozhanggui/system/cashierservice/netty/PushToClientChannelHandlerAdapter.java

@@ -2,17 +2,20 @@ package com.chaozhanggui.system.cashierservice.netty;
 
 import com.alibaba.fastjson.JSONObject;
 import com.chaozhanggui.system.cashierservice.netty.config.NettyChannelHandlerAdapter;
-import com.chaozhanggui.system.cashierservice.util.JSONUtil;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
 
-import java.util.HashMap;
+import java.util.*;
-import java.util.Map;
+import java.util.concurrent.*;
-import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
 
 
 /**
@@ -32,7 +35,10 @@ public class PushToClientChannelHandlerAdapter extends NettyChannelHandlerAdapte
      * [ctx, shopId:clientId]
      */
     private static Map<ChannelHandlerContext, String> clientIdMap = new ConcurrentHashMap<>();
-
+    private static Map<String,Queue<JSONObject>> retryQueue = new HashMap<>();
+    private static ConcurrentHashMap<String, AtomicInteger> retryCounts = new ConcurrentHashMap<>();
+    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(3);
+    int maxRetryAttempts = 5;
 
     private String clientId = "";
     private String shopId = "";
@@ -51,8 +57,8 @@ public class PushToClientChannelHandlerAdapter extends NettyChannelHandlerAdapte
     }
 
     @Override
-    public void channelInactive(ChannelHandlerContext ctx)  {
+    public void channelInactive(ChannelHandlerContext ctx) {
-        log.info("netty连接client 长连接关闭:{}, {}",clientId,shopId);
+        log.info("netty连接client 长连接关闭:{}, {}", clientId, shopId);
         ctx.close();
         removeCtx(ctx);
     }
@@ -68,9 +74,6 @@ public class PushToClientChannelHandlerAdapter extends NettyChannelHandlerAdapte
             ConcurrentHashMap<String, ChannelHandlerContext> tableMap = webSocketMap.get(split[0]);
             if (tableMap != null && !tableMap.isEmpty() && tableMap.size() > 0) {
                 tableMap.remove(split[1]);
-                if (tableMap.isEmpty() || tableMap.size() == 0) {
-                    webSocketMap.remove(split[0]);
-                }
             }
         }
         clientIdMap.remove(ctx);
@@ -92,43 +95,49 @@ public class PushToClientChannelHandlerAdapter extends NettyChannelHandlerAdapte
         JSONObject jsonObject = new JSONObject();
         if (StringUtils.isNotEmpty(msg)) {
             jsonObject = JSONObject.parseObject(msg);
-        }else {
+        } else {
-            log.info("netty连接client 接收到空数据：{}",msg);
+            log.info("netty连接client 接收到空数据：{}", msg);
         }
         String type = jsonObject.getString("type");
-        if(type.equals("heartbeat")){//心跳
+        if (type.equals("heartbeat")) {//心跳
-            log.info("netty连接client 接收到心跳数据：shop:{} clientId:{} meg:{}",shopId,clientId,msg);
+            log.info("netty连接client 接收到心跳数据：shop:{} clientId:{} meg:{}", shopId, clientId, msg);
-        }else {
+        } else {
             if (type.equals("connect")) {
                 String clientId = jsonObject.getString("clientId");
                 String shopId = jsonObject.getString("shopId");
-                if (StringUtils.isBlank(type)  || StringUtils.isBlank(shopId) || StringUtils.isBlank(clientId)) {
+                if (StringUtils.isBlank(type) || StringUtils.isBlank(shopId) || StringUtils.isBlank(clientId)) {
-                    log.info("netty连接client 建立连接请求失败:{}",jsonObject);
+                    log.info("netty连接client 建立连接请求失败:{}", jsonObject);
                     channelInactive(ctx);
                     return;
                 }
-
+                log.info("netty连接client 接收到数据 建立连接参数 param:{}", jsonObject);
-                log.info("netty连接client 接收到数据 建立连接参数 param:{}",jsonObject);
+                this.clientId = clientId;
-                this.clientId=clientId;
+                this.shopId = shopId;
-                this.shopId=shopId;
+                webSocketMap.computeIfAbsent(shopId, k -> new ConcurrentHashMap<>()).put(clientId, ctx);
-                if (webSocketMap.containsKey(shopId)) {
-                    ConcurrentHashMap<String, ChannelHandlerContext> clientSocketMap = webSocketMap.get(shopId);
-                    ChannelHandlerContext channelHandlerContext = clientSocketMap.get(clientId);
-                    if (channelHandlerContext != null) {
-                        channelHandlerContext.close();
-                    }
-                    clientSocketMap.put(clientId, ctx);
-                } else {
-                    ConcurrentHashMap<String, ChannelHandlerContext> clientSocketMap = new ConcurrentHashMap<>();
-                    clientSocketMap.put(clientId, ctx);
-                    webSocketMap.put(shopId,clientSocketMap);
-                }
                 clientIdMap.put(ctx, shopId + ":" + clientId);
                 JSONObject jsonObject1 = new JSONObject();
                 jsonObject1.put("status", "success");
                 jsonObject1.put("msg", "连接成功");
                 jsonObject1.put("type", "connect");
                 sendMesToApp(jsonObject1.toString(), ctx);
+            }else if(type.equals("send")){
+                String orderNo = jsonObject.getString("orderNo");
+//                log.info("netty连接client 接收到send 数据 param:{} orderNo:{} shopId:{}", jsonObject,orderNo,shopId);
+                if (retryQueue.containsKey(shopId)) {
+                    // 获取内层Map
+                    Queue<JSONObject> queue = retryQueue.get(shopId);
+                    if (queue != null) {
+                        Iterator<JSONObject> iterator = queue.iterator();
+                        while (iterator.hasNext()) {
+                            JSONObject customObject = iterator.next();
+//                            log.info("netty连接client 接收到send 数据 retryQueue数据：{}", customObject);
+                            if (customObject.getJSONObject("orderInfo").getString("orderNo").equals(orderNo)) {
+                                iterator.remove();
+                                retryCounts.remove(orderNo);
+                            }
+                        }
+                    }
+                }
             }
         }
         //业务逻辑代码处理框架。。。
@@ -139,53 +148,96 @@ public class PushToClientChannelHandlerAdapter extends NettyChannelHandlerAdapte
         sendMessage(ctx, str);
     }
 
-    /**
-     * @param message 发送的消息内容
-     * @param shopId  店铺Id
-     * @param clientId 客户端Id
-     * @param userFlag
-     *            为true 单发给clientId
-     *            为false 群发 shopId为空 发给所有人
-     */
     @Async
-    public void AppSendInfo(String message, String shopId,String clientId, boolean userFlag) {
+    public void sendMesToApp(String shopId,String orderNo,JSONObject str, ChannelHandlerContext ctx) {
-        log.info("netty连接client 发送消息 shopId:{}  clientId:{} userFlag:{} message:{}",shopId,clientId,userFlag, JSONUtil.toJSONString(message));
+        ctx.channel().writeAndFlush(new TextWebSocketFrame(str.toString())).addListener((GenericFutureListener<Future<? super Void>>) future -> {
-        if (userFlag) {
+            if (!future.isSuccess()) {
-            if (webSocketMap.containsKey(shopId)) {
+                if (shopId != null) {
-                ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(shopId);
+                    retryQueue.computeIfAbsent(shopId, k -> new ConcurrentLinkedQueue<>()).offer(str);
-                if(!webSockets.isEmpty()){
+                    scheduleRetry(shopId,orderNo);
-                    if (StringUtils.isNotBlank(clientId)) {
-                        ChannelHandlerContext ctx = webSockets.get(clientId);
-                        if (ctx != null) {
-                            sendMesToApp(message,ctx);
-                        }
-                    }
                 }
             }
-        } else {
+        });
-            if (StringUtils.isEmpty(shopId)) {
+    }
-                // 向所有用户发送信息
+
-                for (ConcurrentHashMap<String, ChannelHandlerContext> value : webSocketMap.values()) {
+    /**
-                    for (ChannelHandlerContext ctx : value.values()) {
+     * @param message  发送的消息内容
-                        sendMesToApp(message,ctx);
+     * @param shopId   店铺Id
-                    }
+    //     * @param clientId 客户端Id
-                }
+    //     * @param userFlag 为true 单发给clientId
-            } else if (webSocketMap.containsKey(shopId)) {
+     *                 为false 群发 shopId为空 发给所有人
-                ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(shopId);
+     */
-                if(!webSockets.isEmpty()) {
+//    @Async
-                    for (String user : webSockets.keySet()) {
+//    public void AppSendInfo(String message, String shopId,String clientId, boolean userFlag) {
-                        ChannelHandlerContext ctx = webSockets.get(user);
+//        log.info("netty连接client 发送消息 shopId:{}  clientId:{} userFlag:{} message:{}",shopId,clientId,userFlag, JSONUtil.toJSONString(message));
-                        if (ctx != null) {
+//        if (userFlag) {
-                            log.info("netty连接client 发送消息 桌码群发 clientId:{}",user);
+//            if (webSocketMap.containsKey(shopId)) {
-                            sendMesToApp(message,ctx);
+//                ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(shopId);
-                        }else {
+//                if(!webSockets.isEmpty()){
-                            log.info("netty连接client 发送消息 桌码群发 clientId:{} 失败",user);
+//                    if (StringUtils.isNotBlank(clientId)) {
-                        }
+//                        ChannelHandlerContext ctx = webSockets.get(clientId);
-                    }
+//                        if (ctx != null) {
-                }else {
+//                            sendMesToApp(message,ctx);
-                    log.info("netty连接client 发送消息 桌码群发 clientId:{} 失败",clientId);
+//                        }
-                }
+//                    }
+//                }
+//            }
+//        } else {
+//            if (StringUtils.isEmpty(shopId)) {
+//                // 向所有用户发送信息
+//                for (ConcurrentHashMap<String, ChannelHandlerContext> value : webSocketMap.values()) {
+//                    for (ChannelHandlerContext ctx : value.values()) {
+//                        sendMesToApp(message,ctx);
+//                    }
+//                }
+//            } else if (webSocketMap.containsKey(shopId)) {
+//                ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(shopId);
+//                if((!webSockets.isEmpty() && !webSockets.keySet().isEmpty())) {
+//                    for (String user : webSockets.keySet()) {
+//                        ChannelHandlerContext ctx = webSockets.get(user);
+//                        if (ctx != null) {
+//                            log.info("netty连接client 发送消息 桌码群发 clientId:{}",user);
+//                            sendMesToApp(message,ctx);
+//                        }else {
+//                            log.info("netty连接client 发送消息 桌码群发 clientId:{} 失败",user);
+//                        }
+//                    }
+//                }else {
+//                    log.info("netty连接client 发送消息 桌码群发 clientId:{} 失败",clientId);
+//                }
+//            }
+//        }
+//    }
+    public void AppSendInfoV1(String shopId, String orderNo, JSONObject message) {
+        log.info("netty连接client 发送消息 shopId:{}  clientId:{} userFlag:{} message:{}", shopId, message.get("orderInfo"));
+        retryQueue.computeIfAbsent(shopId, k -> new ConcurrentLinkedQueue<>()).offer(message);
+        scheduleRetry(shopId, orderNo);
+        ConcurrentHashMap<String, ChannelHandlerContext> webSockets = webSocketMap.get(shopId);
+        if (webSockets != null) {
+            for (ChannelHandlerContext ctx : webSockets.values()) {
+                sendMesToApp(shopId, orderNo, message, ctx);
+                log.info("netty连接client 向：{}发送消息", shopId);
             }
         }
     }
+
+    // 定时重试方法
+    private void scheduleRetry(String shopId,String orderNo) {
+//        log.info("定时重发");
+        scheduler.schedule(() -> {
+            Queue<JSONObject> shopRetryQueue = retryQueue.get(shopId);
+            if (shopRetryQueue != null) {
+                JSONObject message = shopRetryQueue.poll();
+                if (message != null) {
+                    AtomicInteger retryCount = retryCounts.computeIfAbsent(orderNo, k -> new AtomicInteger(0));
+                    int currentRetryCount = retryCount.incrementAndGet();
+                    if (currentRetryCount <= maxRetryAttempts) {
+                        AppSendInfoV1(shopId,orderNo,message);
+                    } else {
+                        log.error("重试次数超过最大限制，放弃重试。shopId: {}, orderNo: {}", shopId, orderNo);
+                    }
+                }
+            }
+        }, 4, TimeUnit.SECONDS);
+    }
 }

src/main/java/com/chaozhanggui/system/cashierservice/service/CartService.java

@@ -796,7 +796,8 @@ public class CartService {
             jsonObject12.put("amount", BigDecimal.ZERO);
 
             jsonObject12.put("data", new JSONArray());
-            PushToAppChannelHandlerAdapter.getInstance().AppSendInfo(jsonObject12.toString(), jsonObject.getString("tableId").concat("-").concat(shopId), "", false);
+//            PushToAppChannelHandlerAdapter.getInstance().AppSendInfo(jsonObject12.toString(), jsonObject.getString("tableId").concat("-").concat(shopId), "", false);
+            PushToAppChannelHandlerAdapter.getInstance().AppSendInfo(jsonObject12.toString(), jsonObject.getString("tableId").concat("-").concat(shopId), jsonObject.getString("userId"));
             redisUtil.saveMessage(RedisCst.ORDER_EXPIRED.concat(orderId.toString()), orderId.toString(), 60 * 16L);
 
 

src/main/java/com/chaozhanggui/system/cashierservice/service/PayService.java

@@ -1208,7 +1208,7 @@ public class PayService {
 //                e.printStackTrace();
 //            }
 //        }
-        PushToClientChannelHandlerAdapter.getInstance().AppSendInfo(client.toString(), orderInfo.getShopId(), "", false);
+        PushToClientChannelHandlerAdapter.getInstance().AppSendInfoV1(orderInfo.getShopId(),orderInfo.getOrderNo(), client);
     }