diff --git a/client-adapter/launcher/src/main/bin/startup.sh b/client-adapter/launcher/src/main/bin/startup.sh index 67bad9d3..71b817f4 100644 --- a/client-adapter/launcher/src/main/bin/startup.sh +++ b/client-adapter/launcher/src/main/bin/startup.sh @@ -81,12 +81,12 @@ if [ -n "$OS_ARCH" ]; then # For G1 JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS" else - # JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS" + JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS" ## JDK 1.8 2c4g [-XX:ConcGCThreads=1 -XX:ParallelGCThreads=2], 4c8g [-XX:ConcGCThreads=2 -XX:ParallelGCThreads=4] - JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0" - JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4" - JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" - JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m " + #JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0" + #JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4" + #JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" + #JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m " fi else JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS" diff --git a/deployer/src/main/bin/startup.sh b/deployer/src/main/bin/startup.sh index 7f893fcd..ffc96135 100644 --- a/deployer/src/main/bin/startup.sh +++ b/deployer/src/main/bin/startup.sh @@ -106,12 +106,12 @@ if [ -n "$OS_ARCH" ]; then # For G1 JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS" else - # JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS" + JAVA_OPTS="-server -Xms2g -Xmx3g -Xmn1g -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC $JAVA_OPTS" ## JDK 1.8 2c4g [-XX:ConcGCThreads=1 -XX:ParallelGCThreads=2], 4c8g [-XX:ConcGCThreads=2 -XX:ParallelGCThreads=4] - JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0" - JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4" - JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" - JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m " + #JAVA_OPTS="$JAVA_OPTS -XX:InitialRAMPercentage=45.0 -XX:MinRAMPercentage=45.0 -XX:MaxRAMPercentage=45.0" + #JAVA_OPTS="$JAVA_OPTS -XX:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4" + #JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" + #JAVA_OPTS="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m " fi else JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS" diff --git a/server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java index eead88a3..e67fcc79 100644 --- a/server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java @@ -2,14 +2,9 @@ package com.alibaba.otter.canal.admin.handler; import org.apache.commons.lang.exception.ExceptionUtils; import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelStateEvent; -import org.jboss.netty.channel.ExceptionEvent; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelHandler; +import org.jboss.netty.channel.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.slf4j.helpers.MessageFormatter; import com.alibaba.otter.canal.admin.CanalAdmin; import com.alibaba.otter.canal.admin.netty.AdminNettyUtils; @@ -56,7 +51,8 @@ public class SessionHandler extends SimpleChannelHandler { svrResp = AdminNettyUtils.ackPacket(canalAdmin.getRunningInstances()); break; default: - svrResp = AdminNettyUtils.errorPacket(301, "ServerAdmin action: " + sa.getAction() + " is unknown"); + svrResp = AdminNettyUtils.errorPacket(301, + "ServerAdmin action: " + sa.getAction() + " is unknown"); break; } AdminNettyUtils.write(ctx.getChannel(), svrResp); @@ -68,22 +64,28 @@ public class SessionHandler extends SimpleChannelHandler { byte[] instResp = null; switch (ia.getAction()) { case "check": - instResp = AdminNettyUtils.ackPacket(canalAdmin.checkInstance(ia.getDestination()) ? "1" : "0"); + instResp = AdminNettyUtils + .ackPacket(canalAdmin.checkInstance(ia.getDestination()) ? "1" : "0"); break; case "start": - instResp = AdminNettyUtils.ackPacket(canalAdmin.startInstance(ia.getDestination()) ? "1" : "0"); + instResp = AdminNettyUtils + .ackPacket(canalAdmin.startInstance(ia.getDestination()) ? "1" : "0"); break; case "stop": - instResp = AdminNettyUtils.ackPacket(canalAdmin.stopInstance(ia.getDestination()) ? "1" : "0"); + instResp = AdminNettyUtils + .ackPacket(canalAdmin.stopInstance(ia.getDestination()) ? "1" : "0"); break; case "release": - instResp = AdminNettyUtils.ackPacket(canalAdmin.releaseInstance(ia.getDestination()) ? "1" : "0"); + instResp = AdminNettyUtils + .ackPacket(canalAdmin.releaseInstance(ia.getDestination()) ? "1" : "0"); break; case "restart": - instResp = AdminNettyUtils.ackPacket(canalAdmin.restartInstance(ia.getDestination()) ? "1" : "0"); + instResp = AdminNettyUtils + .ackPacket(canalAdmin.restartInstance(ia.getDestination()) ? "1" : "0"); break; default: - instResp = AdminNettyUtils.errorPacket(301, "InstanceAdmin action: " + ia.getAction() + " is unknown"); + instResp = AdminNettyUtils.errorPacket(301, + "InstanceAdmin action: " + ia.getAction() + " is unknown"); break; } AdminNettyUtils.write(ctx.getChannel(), instResp); @@ -105,40 +107,44 @@ public class SessionHandler extends SimpleChannelHandler { if ("list".equalsIgnoreCase(la.getAction())) { logResp = AdminNettyUtils.ackPacket(canalAdmin.listInstanceLog(la.getDestination())); } else { - logResp = AdminNettyUtils.ackPacket(canalAdmin.instanceLog(la.getDestination(), la.getFile(), la.getCount())); + logResp = AdminNettyUtils.ackPacket( + canalAdmin.instanceLog(la.getDestination(), la.getFile(), la.getCount())); } break; default: - logResp = AdminNettyUtils.errorPacket(301, "LogAdmin type: " + la.getType() + " is unknown"); + logResp = AdminNettyUtils.errorPacket(301, + "LogAdmin type: " + la.getType() + " is unknown"); break; } AdminNettyUtils.write(ctx.getChannel(), logResp); break; default: - logResp = AdminNettyUtils.errorPacket(300, "packet type: " + packet.getType() + " is NOT supported!"); + logResp = AdminNettyUtils.errorPacket(300, + "packet type: " + packet.getType() + " is NOT supported!"); AdminNettyUtils.write(ctx.getChannel(), logResp); break; } } catch (Throwable exception) { - String error = "something goes wrong with channel: " + ctx.getChannel() + ", exception: "+ ExceptionUtils.getStackTrace(exception); + String error = "something goes wrong with channel: " + ctx.getChannel() + ", exception: " + + ExceptionUtils.getStackTrace(exception); AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.errorPacket(400, error)); } } -// public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { -// logger.error("something goes wrong with channel:{}, exception={}", -// ctx.getChannel(), -// ExceptionUtils.getStackTrace(e.getCause())); -// ctx.getChannel().close(); -// } -// -// public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { -// } -// -// public void setCanalAdmin(CanalAdmin canalAdmin) { -// this.canalAdmin = canalAdmin; -// } + public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { + logger.error("something goes wrong with channel:{}, exception={}", + ctx.getChannel(), + ExceptionUtils.getStackTrace(e.getCause())); + ctx.getChannel().close(); + } + + public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + } + + public void setCanalAdmin(CanalAdmin canalAdmin) { + this.canalAdmin = canalAdmin; + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index 5abff977..c6a51e1a 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -6,31 +6,19 @@ import java.util.concurrent.TimeUnit; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.jboss.netty.buffer.ChannelBuffer; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ChannelStateEvent; -import org.jboss.netty.channel.ExceptionEvent; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelHandler; +import org.jboss.netty.channel.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; -import org.slf4j.helpers.MessageFormatter; import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor; import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalPacket; -import com.alibaba.otter.canal.protocol.CanalPacket.ClientAck; -import com.alibaba.otter.canal.protocol.CanalPacket.ClientRollback; -import com.alibaba.otter.canal.protocol.CanalPacket.Get; -import com.alibaba.otter.canal.protocol.CanalPacket.Messages; -import com.alibaba.otter.canal.protocol.CanalPacket.Packet; -import com.alibaba.otter.canal.protocol.CanalPacket.PacketType; -import com.alibaba.otter.canal.protocol.CanalPacket.Sub; -import com.alibaba.otter.canal.protocol.CanalPacket.Unsub; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.CanalPacket.*; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.netty.NettyUtils; import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator; @@ -323,22 +311,22 @@ public class SessionHandler extends SimpleChannelHandler { } } -// public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { -// logger.error("something goes wrong with channel:{}, exception={}", -// ctx.getChannel(), -// ExceptionUtils.getStackTrace(e.getCause())); -// -// ctx.getChannel().close(); -// } + public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { + logger.error("something goes wrong with channel:{}, exception={}", + ctx.getChannel(), + ExceptionUtils.getStackTrace(e.getCause())); -// public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { -// // logger.info("remove binding subscription value object if any..."); -// // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment(); -// // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费 -// // if (clientIdentity != null) { -// // stopCanalInstanceIfNecessary(clientIdentity); -// // } -// } + ctx.getChannel().close(); + } + + public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { + // logger.info("remove binding subscription value object if any..."); + // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment(); + // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费 + // if (clientIdentity != null) { + // stopCanalInstanceIfNecessary(clientIdentity); + // } + } private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) { List clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination()); @@ -371,8 +359,8 @@ public class SessionHandler extends SimpleChannelHandler { } } -// public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { -// this.embeddedServer = embeddedServer; -// } + public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { + this.embeddedServer = embeddedServer; + } }