心跳调整

mpush-core/src/main/java/com/shinemo/mpush/core/server/ConnectionServer.java

@@ -1,16 +1,12 @@
 package com.shinemo.mpush.core.server;
 
-import java.util.concurrent.TimeUnit;
 
 import com.shinemo.mpush.api.connection.ConnectionManager;
 import com.shinemo.mpush.api.protocol.Command;
 import com.shinemo.mpush.common.MessageDispatcher;
 import com.shinemo.mpush.core.handler.*;
 import com.shinemo.mpush.netty.connection.NettyConnectionManager;
 import com.shinemo.mpush.netty.server.NettyServer;
-import com.shinemo.mpush.netty.server.ScanAllConnectionTimerTask;
-import com.shinemo.mpush.netty.util.NettySharedHolder;
-import com.shinemo.mpush.tools.config.ConfigCenter;
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelHandler;
@@ -26,7 +22,7 @@ public final class ConnectionServer extends NettyServer {
 
     public ConnectionServer(int port) {
         super(port);
-        NettySharedHolder.HASHED_WHEEL_TIMER.newTimeout(new ScanAllConnectionTimerTask(connectionManager), ConfigCenter.holder.scanConnTaskCycle() / 1000, TimeUnit.SECONDS);
+//        NettySharedHolder.HASHED_WHEEL_TIMER.newTimeout(new ScanAllConnectionTimerTask(connectionManager), ConfigCenter.holder.scanConnTaskCycle() / 1000, TimeUnit.SECONDS);
     }
 
     @Override

mpush-netty/src/main/java/com/shinemo/mpush/netty/connection/NettyConnectionManager.java

@@ -2,24 +2,42 @@
 
 
 import com.google.common.collect.Lists;
+import com.google.common.eventbus.Subscribe;
 import com.shinemo.mpush.api.connection.Connection;
 import com.shinemo.mpush.api.connection.ConnectionManager;
+import com.shinemo.mpush.api.event.HandshakeEvent;
+import com.shinemo.mpush.common.EventBus;
+import com.shinemo.mpush.log.LogType;
+import com.shinemo.mpush.log.LoggerManage;
+import com.shinemo.mpush.tools.config.ConfigCenter;
 
 import io.netty.channel.Channel;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timeout;
+import io.netty.util.Timer;
+import io.netty.util.TimerTask;
 import io.netty.util.internal.chmv8.ConcurrentHashMapV8;
 
 import java.util.List;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Created by ohun on 2015/12/22.
  */
-public final class NettyConnectionManager implements ConnectionManager {
+public final class NettyConnectionManager  implements ConnectionManager {
     //可能会有20w的链接数
     private final ConcurrentMap<String, Connection> connections = new ConcurrentHashMapV8<>();
 
+    private Timer wheelTimer;
+
     @Override
     public void init() {
+    	//每秒钟走一步，一个心跳周期内走一圈
+        long tickDuration = 1000;//1s
+        int ticksPerWheel = (int) (ConfigCenter.holder.maxHeartbeat() / tickDuration);
+        this.wheelTimer = new HashedWheelTimer(tickDuration, TimeUnit.MILLISECONDS, ticksPerWheel);
+        EventBus.INSTANCE.register(this);
     }
 
     @Override
@@ -44,5 +62,44 @@ public void remove(Channel channel) {
     public List<Connection> getConnections() {
     	return Lists.newArrayList(connections.values());
     }
+
+    @Subscribe
+    void onHandshakeOk(HandshakeEvent event) {
+        HeartbeatCheckTask task = new HeartbeatCheckTask(event.heartbeat, event.connection);
+        task.startTimeout();
+    }
+
+    private class HeartbeatCheckTask implements TimerTask {
+        private int expiredTimes = 0;
+        private final int heartbeat;
+        private final Connection connection;
+
+        public HeartbeatCheckTask(int heartbeat, Connection connection) {
+            this.heartbeat = heartbeat;
+            this.connection = connection;
+        }
+
+        public void startTimeout() {
+            wheelTimer.newTimeout(this, heartbeat, TimeUnit.MILLISECONDS);
+        }
+
+        @Override
+        public void run(Timeout timeout) throws Exception {
+            if (!connection.isConnected()) return;
+            if (connection.heartbeatTimeout()) {
+                if (++expiredTimes > ConfigCenter.holder.maxHBTimeoutTimes()) {
+                    connection.close();
+                    LoggerManage.info(LogType.HEARTBEAT, "connection heartbeat timeout, connection has bean closed:%s,%s", connection.getChannel(),connection.getSessionContext().deviceId);
+                    return;
+                } else {
+                	LoggerManage.info(LogType.HEARTBEAT, "connection heartbeat timeout, expiredTimes:%s,%s,%s", expiredTimes,connection.getChannel(),connection.getSessionContext().deviceId);
+                }
+            } else {
+            	expiredTimes = 0;
+            	LoggerManage.info(LogType.HEARTBEAT, "connection heartbeat reset, expiredTimes:%s,%s,%s", expiredTimes,connection.getChannel(),connection.getSessionContext().deviceId);
+            }
+            startTimeout();
+        }
+    }
 
 }

mpush-netty/src/main/java/com/shinemo/mpush/netty/server/NettyServer.java

@@ -28,7 +28,7 @@ public abstract class NettyServer implements Server {
 
     public enum State {Created, Initialized, Starting, Started, Shutdown}
 
-    protected final AtomicReference<State> serverState = new AtomicReference(State.Created);
+	protected final AtomicReference<State> serverState = new AtomicReference<>(State.Created);
 
     private final int port;
     private EventLoopGroup bossGroup;
@@ -162,7 +162,8 @@ private void createNioServer(final Listener listener) {
     }
 
 
-    private void createEpollServer(final Listener listener) {
+    @SuppressWarnings("unused")
+	private void createEpollServer(final Listener listener) {
         EpollEventLoopGroup bossGroup = new EpollEventLoopGroup(1, ThreadPoolManager.bossExecutor);
         EpollEventLoopGroup workerGroup = new EpollEventLoopGroup(0, ThreadPoolManager.workExecutor);
         createServer(listener, bossGroup, workerGroup, EpollServerSocketChannel.class);

mpush-test/src/test/java/com/shinemo/mpush/test/redis/PubSubTest.java

@@ -1,6 +1,7 @@
 package com.shinemo.mpush.test.redis;
 
 import java.util.List;
+import java.util.concurrent.locks.LockSupport;
 
 import org.junit.Before;
 import org.junit.Test;
@@ -27,14 +28,32 @@ public void init(){
     }
 
 	@Test
-	public void pubSubTest(){
-
+	public void subpubTest(){
 		RedisManage.subscribe(Subscriber.holder, "/hello/123");
-
-		RedisManage.subscribe(Subscriber.holder, "/hello/124");
-
+		RedisManage.subscribe(Subscriber.holder, "/hello/124");	
 		RedisManage.publish("/hello/123", "123");
 		RedisManage.publish("/hello/124", "124");
 	}
 
+	@Test
+	public void pubsubTest(){
+		RedisManage.publish("/hello/123", "123");
+		RedisManage.publish("/hello/124", "124");
+		RedisManage.subscribe(Subscriber.holder, "/hello/123");
+		RedisManage.subscribe(Subscriber.holder, "/hello/124");	
+	}
+
+	@Test
+	public void pubTest(){
+		RedisManage.publish("/hello/123", "123");
+		RedisManage.publish("/hello/124", "124");
+	}
+
+	@Test
+	public void subTest(){
+		RedisManage.subscribe(Subscriber.holder, "/hello/123");
+		RedisManage.subscribe(Subscriber.holder, "/hello/124");	
+		LockSupport.park();
+	}
+
 }

mpush-test/src/test/resources/logback.xml

@@ -15,7 +15,7 @@
 		<target>System.err</target>
 		<encoding>UTF-8</encoding>
 		<filter class="ch.qos.logback.classic.filter.ThresholdFilter">  
-            <level>WARN</level>  
+            <level>debug</level>  
         </filter>
 		<layout class="ch.qos.logback.classic.PatternLayout">
 			<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} - [%thread] %-5level - %logger{35}  - %msg%n</pattern>
@@ -24,7 +24,7 @@
 
 
 	<root>
-		<level value="warn" />
+		<level value="debug" />
 		<appender-ref ref="STDOUT" />
 	</root>
 </configuration>