diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java new file mode 100644 index 00000000..b3d320b1 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java @@ -0,0 +1,18 @@ +package me.chanjar.weixin.common.util; + +/** + *
+ * 消息重复检查器
+ * 微信服务器在五秒内收不到响应会断掉连接,并且重新发起请求,总共重试三次
+ * 
+ */ +public interface WxMsgIdDuplicateChecker { + + /** + * 检查消息ID是否重复 + * @param wxMsgId + * @return 如果是重复消息,返回true,否则返回false + */ + public boolean isDuplicate(Long wxMsgId); + +} diff --git a/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java new file mode 100644 index 00000000..a7bb70c6 --- /dev/null +++ b/weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java @@ -0,0 +1,84 @@ +package me.chanjar.weixin.common.util; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + *
+ * 默认消息重复检查器
+ * 将每个消息id保存在内存里,每隔5秒清理已经过期的消息id,每个消息id的过期时间是15秒
+ * 
+ */ +public class WxMsgIdInMemoryDuplicateChecker implements WxMsgIdDuplicateChecker { + + /** + * 一个消息ID在内存的过期时间:15秒 + */ + private final Long TIME_TO_LIVE; + + /** + * 每隔多少周期检查消息ID是否过期:5秒 + */ + private final Long CLEAR_PERIOD; + + private final ConcurrentHashMap msgId2Timestamp = new ConcurrentHashMap(); + + /** + * WxMsgIdInMemoryDuplicateChecker构造函数 + *
+   * 一个消息ID在内存的过期时间:15秒
+   * 每隔多少周期检查消息ID是否过期:5秒
+   * 
+ */ + public WxMsgIdInMemoryDuplicateChecker() { + this.TIME_TO_LIVE = 15 * 1000l; + this.CLEAR_PERIOD = 5 * 1000l; + this.start(); + } + + /** + * WxMsgIdInMemoryDuplicateChecker构造函数 + * @param timeToLive 一个消息ID在内存的过期时间:毫秒 + * @param clearPeriod 每隔多少周期检查消息ID是否过期:毫秒 + */ + public WxMsgIdInMemoryDuplicateChecker(Long timeToLive, Long clearPeriod) { + this.TIME_TO_LIVE = timeToLive; + this.CLEAR_PERIOD = clearPeriod; + this.start(); + } + + private void start() { + Thread t = new Thread(new Runnable() { + @Override + public void run() { + try { + while (true) { + Thread.sleep(CLEAR_PERIOD); + Long now = System.currentTimeMillis(); + for (Map.Entry entry : msgId2Timestamp.entrySet()) { + if (now - entry.getValue() > TIME_TO_LIVE) { + msgId2Timestamp.entrySet().remove(entry); + } + } + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + } + + @Override + public boolean isDuplicate(Long wxMsgId) { + Long timestamp = msgId2Timestamp.putIfAbsent(wxMsgId, System.currentTimeMillis()); + if (timestamp == null) { + // 第一次接收到这个消息 + return false; + } + return true; + } + + +} diff --git a/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java b/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java new file mode 100644 index 00000000..a35eba5c --- /dev/null +++ b/weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java @@ -0,0 +1,38 @@ +package me.chanjar.weixin.common.util; + +import org.testng.Assert; +import org.testng.annotations.Test; + +/** + * Created by qianjia on 15/1/20. + */ +@Test +public class WxMsgIdInMemoryDuplicateCheckerTest { + + public void test() throws InterruptedException { + Long[] msgIds = new Long[] { 1l, 2l, 3l, 4l, 5l, 6l, 7l, 8l }; + WxMsgIdInMemoryDuplicateChecker checker = new WxMsgIdInMemoryDuplicateChecker(2000l, 1000l); + + // 第一次检查 + for (Long msgId : msgIds) { + boolean result = checker.isDuplicate(msgId); + Assert.assertFalse(result); + } + + // 过1秒再检查 + Thread.sleep(1000l); + for (Long msgId : msgIds) { + boolean result = checker.isDuplicate(msgId); + Assert.assertTrue(result); + } + + // 过1.5秒再检查 + Thread.sleep(1500l); + for (Long msgId : msgIds) { + boolean result = checker.isDuplicate(msgId); + Assert.assertFalse(result); + } + + } + +} diff --git a/weixin-java-common/src/test/resources/testng.xml b/weixin-java-common/src/test/resources/testng.xml index f0058b53..49d65248 100644 --- a/weixin-java-common/src/test/resources/testng.xml +++ b/weixin-java-common/src/test/resources/testng.xml @@ -7,6 +7,7 @@ + diff --git a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java index 5b4e1820..c88bc8a3 100644 --- a/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java +++ b/weixin-java-cp/src/main/java/me/chanjar/weixin/cp/api/WxCpMessageRouter.java @@ -1,5 +1,7 @@ package me.chanjar.weixin.cp.api; +import me.chanjar.weixin.common.util.WxMsgIdDuplicateChecker; +import me.chanjar.weixin.common.util.WxMsgIdInMemoryDuplicateChecker; import me.chanjar.weixin.cp.bean.WxCpXmlMessage; import me.chanjar.weixin.cp.bean.WxCpXmlOutMessage; @@ -45,18 +47,38 @@ public class WxCpMessageRouter { private final List rules = new ArrayList(); - private final ExecutorService executorService; - private final WxCpService wxCpService; + private ExecutorService executorService; + + private WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker; + public WxCpMessageRouter(WxCpService wxCpService) { this.wxCpService = wxCpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); + this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); } public WxCpMessageRouter(WxCpService wxMpService, int threadPoolSize) { this.wxCpService = wxMpService; this.executorService = Executors.newFixedThreadPool(threadPoolSize); + this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); + } + + /** + * 设置自定义的ExecutorService + * @param executorService + */ + public void setExecutorService(ExecutorService executorService) { + this.executorService = executorService; + } + + /** + * 设置自定义的WxMsgIdDuplicateChecker + * @param wxMsgIdDuplicateChecker + */ + public void setWxMsgIdDuplicateChecker(WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker) { + this.wxMsgIdDuplicateChecker = wxMsgIdDuplicateChecker; } /** @@ -72,6 +94,11 @@ public class WxCpMessageRouter { * @param wxMessage */ public WxCpXmlOutMessage route(final WxCpXmlMessage wxMessage) { + if (wxMsgIdDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + // 如果是重复消息,那么就不做处理 + return null; + } + final List matchRules = new ArrayList(); // 收集匹配的规则 for (final Rule rule : rules) { diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java index e904e41b..7fcc3c96 100644 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpMessageRouter.java @@ -1,5 +1,7 @@ package me.chanjar.weixin.mp.api; +import me.chanjar.weixin.common.util.WxMsgIdDuplicateChecker; +import me.chanjar.weixin.common.util.WxMsgIdInMemoryDuplicateChecker; import me.chanjar.weixin.mp.bean.WxMpXmlMessage; import me.chanjar.weixin.mp.bean.WxMpXmlOutMessage; @@ -45,18 +47,38 @@ public class WxMpMessageRouter { private final List rules = new ArrayList(); - private final ExecutorService executorService; - private final WxMpService wxMpService; + private ExecutorService executorService; + + private WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker; + public WxMpMessageRouter(WxMpService wxMpService) { this.wxMpService = wxMpService; this.executorService = Executors.newFixedThreadPool(DEFAULT_THREAD_POOL_SIZE); + this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); } public WxMpMessageRouter(WxMpService wxMpService, int threadPoolSize) { this.wxMpService = wxMpService; this.executorService = Executors.newFixedThreadPool(threadPoolSize); + this.wxMsgIdDuplicateChecker = new WxMsgIdInMemoryDuplicateChecker(); + } + + /** + * 设置自定义的ExecutorService + * @param executorService + */ + public void setExecutorService(ExecutorService executorService) { + this.executorService = executorService; + } + + /** + * 设置自定义的WxMsgIdDuplicateChecker + * @param wxMsgIdDuplicateChecker + */ + public void setWxMsgIdDuplicateChecker(WxMsgIdDuplicateChecker wxMsgIdDuplicateChecker) { + this.wxMsgIdDuplicateChecker = wxMsgIdDuplicateChecker; } /** @@ -72,6 +94,11 @@ public class WxMpMessageRouter { * @param wxMessage */ public WxMpXmlOutMessage route(final WxMpXmlMessage wxMessage) { + if (wxMsgIdDuplicateChecker.isDuplicate(wxMessage.getMsgId())) { + // 如果是重复消息,那么就不做处理 + return null; + } + final List matchRules = new ArrayList(); // 收集匹配的规则 for (final Rule rule : rules) {