🎨 添加redis微信消息重复检查相关类

Author: biubiubiu3971

weixin-java-common/src/main/java/me/chanjar/weixin/common/api/WxMessageInRedisDuplicateChecker.java

@@ -0,0 +1,47 @@
+package me.chanjar.weixin.common.api;
+
+import lombok.RequiredArgsConstructor;
+import org.redisson.api.RBucket;
+import org.redisson.api.RedissonClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 利用redis检查消息是否重复
+ *
+ */
+@RequiredArgsConstructor
+public class WxMessageInRedisDuplicateChecker implements WxMessageDuplicateChecker {
+
+ /**
+ * 过期时间
+ */
+ private int expire = 10;
+
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ private final RedissonClient redissonClient;
+
+ /**
+ * messageId是否重复
+ *
+ * @param messageId messageId
+ * @return 是否
+ */
+ @Override
+ public boolean isDuplicate(String messageId) {
+ RBucket<String> r = redissonClient.getBucket("wx:message:duplicate:check:" + messageId);
+ boolean setSuccess = r.trySet("1", expire, TimeUnit.SECONDS);
+ return !setSuccess;
+ }
+
+ public int getExpire() {
+ return expire;
+ }
+
+ public void setExpire(int expire) {
+ this.expire = expire;
+ }
+}

weixin-java-common/src/test/java/me/chanjar/weixin/common/api/WxMessageInRedisDuplicateCheckerTest.java

@@ -0,0 +1,57 @@
+package me.chanjar.weixin.common.api;
+
+import org.redisson.Redisson;
+import org.redisson.api.RedissonClient;
+import org.redisson.config.Config;
+import org.redisson.config.TransportMode;
+import org.testng.annotations.BeforeTest;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+@Test
+public class WxMessageInRedisDuplicateCheckerTest {
+
+ private RedissonClient redissonClient;
+
+ @BeforeTest
+ public void init() {
+ Config config = new Config();
+ config.useSingleServer().setAddress("redis://127.0.0.1:6379");
+ config.setTransportMode(TransportMode.NIO);
+ this.redissonClient = Redisson.create(config);
+ checker = new WxMessageInRedisDuplicateChecker(redissonClient);
+ checker.setExpire(2);
+ }
+
+ private WxMessageInRedisDuplicateChecker checker;
+
+ public void test() throws InterruptedException {
+ Long[] msgIds = new Long[]{1L, 2L, 3L, 4L, 5L, 6L, 7L, 8L};
+
+ // 第一次检查
+ for (Long msgId : msgIds) {
+ boolean result = checker.isDuplicate(String.valueOf(msgId));
+ assertFalse(result);
+ }
+
+ // 过1秒再检查
+ TimeUnit.SECONDS.sleep(1);
+ for (Long msgId : msgIds) {
+ boolean result = checker.isDuplicate(String.valueOf(msgId));
+ assertTrue(result);
+ }
+
+ // 过1.5秒再检查
+ TimeUnit.MILLISECONDS.sleep(1500L);
+ for (Long msgId : msgIds) {
+ boolean result = checker.isDuplicate(String.valueOf(msgId));
+ assertFalse(result);
+ }
+
+ }
+
+}

weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java

@@ -14,10 +14,12 @@
 import me.chanjar.weixin.common.util.LogExceptionHandler;
 import me.chanjar.weixin.mp.bean.message.WxMpXmlMessage;
 import me.chanjar.weixin.mp.bean.message.WxMpXmlOutMessage;
+import me.chanjar.weixin.mp.util.WxMpConfigStorageHolder;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.xml.ws.Holder;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -208,11 +210,14 @@ public WxMpXmlOutMessage route(final WxMpXmlMessage wxMessage, final Map<String,
 
  WxMpXmlOutMessage res = null;
  final List<Future<?>> futures = new ArrayList<>();
+ String appId = WxMpConfigStorageHolder.get();
  for (final WxMpMessageRouterRule rule : matchRules) {
  // 返回最后一个非异步的rule的执行结果
  if (rule.isAsync()) {
  futures.add(
  this.executorService.submit(() -> {
+ //传入父线程的appId
+ this.wxMpService.switchoverTo(appId);
  rule.service(wxMessage, context, mpService, WxMpMessageRouter.this.sessionManager, WxMpMessageRouter.this.exceptionHandler);
  })
  );