chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

merge code

Browse Source
This commit is contained in:
jianghang.loujh
2026-01-28 12:06:35 +08:00
parent f07a075dce
commit 4ded2869c9
4 changed files with 66 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
client-adapter/launcher/src/main/bin/startup.sh
View File

@@ -81,12 +81,12 @@ if [ -n "$OS_ARCH" ]; then
# For G1 # For G1
JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS" JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
else 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] ## 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: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:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4"
JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" #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="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m "
fi fi
else else
JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS" JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"

10
deployer/src/main/bin/startup.sh
View File

@@ -106,12 +106,12 @@ if [ -n "$OS_ARCH" ]; then
# For G1 # For G1
JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS" JAVA_OPTS="-server -Xms2g -Xmx3g -XX:+UseG1GC -XX:MaxGCPauseMillis=250 -XX:+UseGCOverheadLimit -XX:+ExplicitGCInvokesConcurrent $JAVA_OPTS"
else 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] ## 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: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:-UseParallelGC -XX:+UseConcMarkSweepGC -XX:ConcGCThreads=2 -XX:ParallelGCThreads=4"
JAVA_OPTS="$JAVA_OPTS -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=70" #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="-server -XX:NewRatio=1 -XX:SurvivorRatio=2 -XX:MetaspaceSize=96m -XX:MaxMetaspaceSize=128m "
fi fi
else else
JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS" JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m $JAVA_OPTS"

66
server/src/main/java/com/alibaba/otter/canal/admin/handler/SessionHandler.java
View File

@@ -2,14 +2,9 @@ package com.alibaba.otter.canal.admin.handler;
import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.exception.ExceptionUtils;
import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.*;
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.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import com.alibaba.otter.canal.admin.CanalAdmin; import com.alibaba.otter.canal.admin.CanalAdmin;
import com.alibaba.otter.canal.admin.netty.AdminNettyUtils; import com.alibaba.otter.canal.admin.netty.AdminNettyUtils;
@@ -56,7 +51,8 @@ public class SessionHandler extends SimpleChannelHandler {
svrResp = AdminNettyUtils.ackPacket(canalAdmin.getRunningInstances()); svrResp = AdminNettyUtils.ackPacket(canalAdmin.getRunningInstances());
break; break;
default: default:
svrResp = AdminNettyUtils.errorPacket(301, "ServerAdmin action: " + sa.getAction() + " is unknown"); svrResp = AdminNettyUtils.errorPacket(301,
"ServerAdmin action: " + sa.getAction() + " is unknown");
break; break;
} }
AdminNettyUtils.write(ctx.getChannel(), svrResp); AdminNettyUtils.write(ctx.getChannel(), svrResp);
@@ -68,22 +64,28 @@ public class SessionHandler extends SimpleChannelHandler {
byte[] instResp = null; byte[] instResp = null;
switch (ia.getAction()) { switch (ia.getAction()) {
case "check": case "check":
instResp = AdminNettyUtils.ackPacket(canalAdmin.checkInstance(ia.getDestination()) ? "1" : "0"); instResp = AdminNettyUtils
.ackPacket(canalAdmin.checkInstance(ia.getDestination()) ? "1" : "0");
break; break;
case "start": case "start":
instResp = AdminNettyUtils.ackPacket(canalAdmin.startInstance(ia.getDestination()) ? "1" : "0"); instResp = AdminNettyUtils
.ackPacket(canalAdmin.startInstance(ia.getDestination()) ? "1" : "0");
break; break;
case "stop": case "stop":
instResp = AdminNettyUtils.ackPacket(canalAdmin.stopInstance(ia.getDestination()) ? "1" : "0"); instResp = AdminNettyUtils
.ackPacket(canalAdmin.stopInstance(ia.getDestination()) ? "1" : "0");
break; break;
case "release": case "release":
instResp = AdminNettyUtils.ackPacket(canalAdmin.releaseInstance(ia.getDestination()) ? "1" : "0"); instResp = AdminNettyUtils
.ackPacket(canalAdmin.releaseInstance(ia.getDestination()) ? "1" : "0");
break; break;
case "restart": case "restart":
instResp = AdminNettyUtils.ackPacket(canalAdmin.restartInstance(ia.getDestination()) ? "1" : "0"); instResp = AdminNettyUtils
.ackPacket(canalAdmin.restartInstance(ia.getDestination()) ? "1" : "0");
break; break;
default: default:
instResp = AdminNettyUtils.errorPacket(301, "InstanceAdmin action: " + ia.getAction() + " is unknown"); instResp = AdminNettyUtils.errorPacket(301,
"InstanceAdmin action: " + ia.getAction() + " is unknown");
break; break;
} }
AdminNettyUtils.write(ctx.getChannel(), instResp); AdminNettyUtils.write(ctx.getChannel(), instResp);
@@ -105,40 +107,44 @@ public class SessionHandler extends SimpleChannelHandler {
if ("list".equalsIgnoreCase(la.getAction())) { if ("list".equalsIgnoreCase(la.getAction())) {
logResp = AdminNettyUtils.ackPacket(canalAdmin.listInstanceLog(la.getDestination())); logResp = AdminNettyUtils.ackPacket(canalAdmin.listInstanceLog(la.getDestination()));
} else { } else {
logResp = AdminNettyUtils.ackPacket(canalAdmin.instanceLog(la.getDestination(), la.getFile(), la.getCount())); logResp = AdminNettyUtils.ackPacket(
canalAdmin.instanceLog(la.getDestination(), la.getFile(), la.getCount()));
} }
break; break;
default: default:
logResp = AdminNettyUtils.errorPacket(301, "LogAdmin type: " + la.getType() + " is unknown"); logResp = AdminNettyUtils.errorPacket(301,
"LogAdmin type: " + la.getType() + " is unknown");
break; break;
} }
AdminNettyUtils.write(ctx.getChannel(), logResp); AdminNettyUtils.write(ctx.getChannel(), logResp);
break; break;
default: 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); AdminNettyUtils.write(ctx.getChannel(), logResp);
break; break;
} }
} catch (Throwable exception) { } 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)); AdminNettyUtils.write(ctx.getChannel(), AdminNettyUtils.errorPacket(400, error));
} }
} }
// public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
// logger.error("something goes wrong with channel:{}, exception={}", logger.error("something goes wrong with channel:{}, exception={}",
// ctx.getChannel(), ctx.getChannel(),
// ExceptionUtils.getStackTrace(e.getCause())); ExceptionUtils.getStackTrace(e.getCause()));
// ctx.getChannel().close(); ctx.getChannel().close();
// } }
//
// public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
// } }
//
// public void setCanalAdmin(CanalAdmin canalAdmin) { public void setCanalAdmin(CanalAdmin canalAdmin) {
// this.canalAdmin = canalAdmin; this.canalAdmin = canalAdmin;
// } }
} }

52
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java
View File

@@ -6,31 +6,19 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.exception.ExceptionUtils;
import org.jboss.netty.buffer.ChannelBuffer; import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.*;
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.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.slf4j.MDC; import org.slf4j.MDC;
import org.slf4j.helpers.MessageFormatter;
import org.springframework.util.CollectionUtils; import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor; import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitor;
import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors; import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalPacket; 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.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message; 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.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.NettyUtils; import com.alibaba.otter.canal.server.netty.NettyUtils;
import com.alibaba.otter.canal.server.netty.listener.ChannelFutureAggregator; 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 { public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
// logger.error("something goes wrong with channel:{}, exception={}", logger.error("something goes wrong with channel:{}, exception={}",
// ctx.getChannel(), ctx.getChannel(),
// ExceptionUtils.getStackTrace(e.getCause())); ExceptionUtils.getStackTrace(e.getCause()));
//
// ctx.getChannel().close();
// }
// public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { ctx.getChannel().close();
// // logger.info("remove binding subscription value object if any..."); }
// // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment();
// // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费 public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
// // if (clientIdentity != null) { // logger.info("remove binding subscription value object if any...");
// // stopCanalInstanceIfNecessary(clientIdentity); // ClientIdentity clientIdentity = (ClientIdentity) ctx.getAttachment();
// // } // // 如果唯一的订阅者都取消了订阅，直接关闭服务，针对内部版本模式下可以减少资源浪费
// } // if (clientIdentity != null) {
// stopCanalInstanceIfNecessary(clientIdentity);
// }
}
private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) { private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination()); List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
@@ -371,8 +359,8 @@ public class SessionHandler extends SimpleChannelHandler {
} }
} }
// public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
// this.embeddedServer = embeddedServer; this.embeddedServer = embeddedServer;
// } }
} }