From 85168a14aa9833fa4f9af85c378bbdb1a518f975 Mon Sep 17 00:00:00 2001 From: Daniel Qian Date: Mon, 19 Jan 2015 17:25:41 +0800 Subject: [PATCH 1/2] =?UTF-8?q?issue=20#66=20=E6=B6=88=E6=81=AF=E5=8E=BB?= =?UTF-8?q?=E9=87=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../api/WxMpDuplicateMessageInterceptor.java | 88 +++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java diff --git a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java new file mode 100644 index 00000000..970bbba4 --- /dev/null +++ b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java @@ -0,0 +1,88 @@ +package me.chanjar.weixin.mp.api; + +import me.chanjar.weixin.mp.bean.WxMpXmlMessage; +import sun.applet.Main; + +import java.net.SocketTimeoutException; +import java.util.Map; +import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; + +/** + *
+ * 消息去重拦截器
+ * 微信服务器在五秒内收不到响应会断掉连接,并且重新发起请求,总共重试三次
+ * 使用方法:
+ * WxMpMessageRouter router = new WxMpMessageRouter();
+ * router
+ *   .rule()
+ *       .interceptor(new WxMpDuplicateMessageInterceptor())
+ *   .next()
+ *   .rule()
+ *       .msgType("MSG_TYPE").event("EVENT").eventKey("EVENT_KEY").content("CONTENT")
+ *       .interceptor(interceptor, ...).handler(handler, ...)
+ *   .end()
+ *   .rule()
+ *       // 另外一个匹配规则
+ *   .end()
+ * ;
+ * 
+ */ +public class WxMpDuplicateMessageInterceptor implements WxMpMessageInterceptor { + + private static final Long PERIOD = 15 * 1000l; + + private final ConcurrentHashMap msgId2timestamp; + + public WxMpDuplicateMessageInterceptor() { + this.msgId2timestamp = new ConcurrentHashMap(); + Thread t = new Thread(new Runnable() { + @Override + public void run() { + try { + while (true) { + Thread.sleep(PERIOD); + Long now = System.currentTimeMillis(); + for (Map.Entry entry : msgId2timestamp.entrySet()) { + if (now - entry.getValue() > PERIOD) { + msgId2timestamp.entrySet().remove(entry); + } + } + msgId2timestamp.clear(); + } + } catch (InterruptedException e) { + e.printStackTrace(); + } + } + }); + t.setDaemon(true); + t.start(); + } + + @Override + public boolean intercept(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService) { + Long now = System.currentTimeMillis(); + Long timestamp = msgId2timestamp.putIfAbsent(wxMessage.getMsgId(), now); + if (timestamp == null) { + return true; + } + if (timestamp.equals(now)) { + // 第一次接收到这个消息 + return true; + } + return false; + } + + public static void main(String[] args) { + WxMpDuplicateMessageInterceptor d = new WxMpDuplicateMessageInterceptor(); + Long endTime = System.currentTimeMillis() + 30 * 1000; + Random r = new Random(); + + while(System.currentTimeMillis() < endTime) { + WxMpXmlMessage m = new WxMpXmlMessage(); + m.setMsgId(r.nextLong() % 100); + d.intercept(m, null, null); + } + + } +} From 9fe73319d1457770d793ca48b5928819a2255dfe Mon Sep 17 00:00:00 2001 From: Daniel Qian Date: Tue, 20 Jan 2015 09:49:39 +0800 Subject: [PATCH 2/2] issue #66 --- .../common/util/WxMsgIdDuplicateChecker.java | 18 ++++ .../util/WxMsgIdInMemoryDuplicateChecker.java | 84 ++++++++++++++++++ .../WxMsgIdInMemoryDuplicateCheckerTest.java | 38 ++++++++ .../src/test/resources/testng.xml | 1 + .../weixin/cp/api/WxCpMessageRouter.java | 31 ++++++- .../api/WxMpDuplicateMessageInterceptor.java | 88 ------------------- .../weixin/mp/api/WxMpMessageRouter.java | 31 ++++++- 7 files changed, 199 insertions(+), 92 deletions(-) create mode 100644 weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdDuplicateChecker.java create mode 100644 weixin-java-common/src/main/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateChecker.java create mode 100644 weixin-java-common/src/test/java/me/chanjar/weixin/common/util/WxMsgIdInMemoryDuplicateCheckerTest.java delete mode 100644 weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java 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/WxMpDuplicateMessageInterceptor.java b/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java deleted file mode 100644 index 970bbba4..00000000 --- a/weixin-java-mp/src/main/java/me/chanjar/weixin/mp/api/WxMpDuplicateMessageInterceptor.java +++ /dev/null @@ -1,88 +0,0 @@ -package me.chanjar.weixin.mp.api; - -import me.chanjar.weixin.mp.bean.WxMpXmlMessage; -import sun.applet.Main; - -import java.net.SocketTimeoutException; -import java.util.Map; -import java.util.Random; -import java.util.concurrent.ConcurrentHashMap; - -/** - *
- * 消息去重拦截器
- * 微信服务器在五秒内收不到响应会断掉连接,并且重新发起请求,总共重试三次
- * 使用方法:
- * WxMpMessageRouter router = new WxMpMessageRouter();
- * router
- *   .rule()
- *       .interceptor(new WxMpDuplicateMessageInterceptor())
- *   .next()
- *   .rule()
- *       .msgType("MSG_TYPE").event("EVENT").eventKey("EVENT_KEY").content("CONTENT")
- *       .interceptor(interceptor, ...).handler(handler, ...)
- *   .end()
- *   .rule()
- *       // 另外一个匹配规则
- *   .end()
- * ;
- * 
- */ -public class WxMpDuplicateMessageInterceptor implements WxMpMessageInterceptor { - - private static final Long PERIOD = 15 * 1000l; - - private final ConcurrentHashMap msgId2timestamp; - - public WxMpDuplicateMessageInterceptor() { - this.msgId2timestamp = new ConcurrentHashMap(); - Thread t = new Thread(new Runnable() { - @Override - public void run() { - try { - while (true) { - Thread.sleep(PERIOD); - Long now = System.currentTimeMillis(); - for (Map.Entry entry : msgId2timestamp.entrySet()) { - if (now - entry.getValue() > PERIOD) { - msgId2timestamp.entrySet().remove(entry); - } - } - msgId2timestamp.clear(); - } - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - }); - t.setDaemon(true); - t.start(); - } - - @Override - public boolean intercept(WxMpXmlMessage wxMessage, Map context, WxMpService wxMpService) { - Long now = System.currentTimeMillis(); - Long timestamp = msgId2timestamp.putIfAbsent(wxMessage.getMsgId(), now); - if (timestamp == null) { - return true; - } - if (timestamp.equals(now)) { - // 第一次接收到这个消息 - return true; - } - return false; - } - - public static void main(String[] args) { - WxMpDuplicateMessageInterceptor d = new WxMpDuplicateMessageInterceptor(); - Long endTime = System.currentTimeMillis() + 30 * 1000; - Random r = new Random(); - - while(System.currentTimeMillis() < endTime) { - WxMpXmlMessage m = new WxMpXmlMessage(); - m.setMsgId(r.nextLong() % 100); - d.intercept(m, null, null); - } - - } -} 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) {