From b29bc6280786e7fa833f2555e24ff64c5c432a0c Mon Sep 17 00:00:00 2001 From: ascrutae Date: Mon, 25 Jan 2016 16:41:21 +0800 Subject: [PATCH] =?UTF-8?q?=E5=B0=86=E5=8F=91=E9=80=81=E7=AB=AF=E6=94=B9?= =?UTF-8?q?=E6=88=90netty=E5=BA=95=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/ai/cloud/skywalking/conf/Config.java | 2 +- .../cloud/skywalking/sender/DataSender.java | 86 +++++++++++-------- .../sender/DataSenderFactoryWithBalance.java | 12 ++- 3 files changed, 62 insertions(+), 38 deletions(-) diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java index 9b5689696..caccd32cc 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/conf/Config.java @@ -50,7 +50,7 @@ public class Config { public static int CONNECT_PERCENT = 50; // 发送服务端配置 - public static String SERVERS_ADDR = "127.0.0.1:34000;127.0.0.1:34001;127.0.0.1:34002"; + public static String SERVERS_ADDR = "127.0.0.1:34000"; // 是否开启发送 public static boolean IS_OFF = false; diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java index 0e1ebbae5..3e2d454dc 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSender.java @@ -1,38 +1,53 @@ package com.ai.cloud.skywalking.sender; +import com.ai.cloud.io.netty.bootstrap.Bootstrap; +import com.ai.cloud.io.netty.channel.*; +import com.ai.cloud.io.netty.channel.nio.NioEventLoopGroup; +import com.ai.cloud.io.netty.channel.socket.ServerSocketChannel; +import com.ai.cloud.io.netty.channel.socket.nio.NioSocketChannel; +import com.ai.cloud.io.netty.handler.codec.LengthFieldBasedFrameDecoder; +import com.ai.cloud.io.netty.handler.codec.LengthFieldPrepender; +import com.ai.cloud.io.netty.handler.codec.bytes.ByteArrayDecoder; +import com.ai.cloud.io.netty.handler.codec.bytes.ByteArrayEncoder; + import java.io.IOException; import java.net.InetSocketAddress; -import java.net.StandardSocketOptions; -import java.nio.ByteBuffer; -import java.nio.channels.SelectionKey; -import java.nio.channels.Selector; -import java.nio.channels.SocketChannel; -import java.util.logging.Level; import java.util.logging.Logger; -import com.ai.cloud.skywalking.sender.protocol.ProtocolBuilder; - -public class DataSender implements IDataSender{ - private static Logger logger = Logger.getLogger(DataSender.class.getName()); - - private SocketChannel socketChannel; - private Selector selector; - private InetSocketAddress socketAddress; +public class DataSender extends ChannelInboundHandlerAdapter implements IDataSender { + private static Logger logger = Logger.getLogger(DataSender.class.getName()); + private EventLoopGroup group; private SenderStatus status = SenderStatus.FAILED; + private InetSocketAddress socketAddress; + private ChannelFuture channelFuture; public DataSender(String ip, int port) throws IOException { - this(new InetSocketAddress(ip, port)); + this(new InetSocketAddress(ip, port)); } public DataSender(InetSocketAddress address) throws IOException { - selector = Selector.open(); - socketChannel = SocketChannel.open(address); - //设置该sc已非阻塞的方式工作 - socketChannel.configureBlocking(false); - socketChannel.register(selector, SelectionKey.OP_CONNECT); - socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true); this.socketAddress = address; status = SenderStatus.READY; + group = new NioEventLoopGroup(); + try { + Bootstrap bootstrap = new Bootstrap(); + bootstrap.group(group) + .channel(NioSocketChannel.class) + .option(ChannelOption.TCP_NODELAY, true) + .handler(new ChannelInitializer() { + @Override + protected void initChannel(ServerSocketChannel ch) throws Exception { + ChannelPipeline p = ch.pipeline(); + p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4)); + p.addLast("frameEncoder", new LengthFieldPrepender(4)); + p.addLast("decoder", new ByteArrayDecoder()); + p.addLast("encoder", new ByteArrayEncoder()); + } + }); + channelFuture = bootstrap.connect(address).sync(); + } catch (Exception e) { + status = SenderStatus.FAILED; + } } /** @@ -43,29 +58,30 @@ public class DataSender implements IDataSender{ */ @Override public boolean send(String data) { - // 发送报文 try { - socketChannel.register(selector, SelectionKey.OP_READ); - socketChannel.write(ByteBuffer.wrap(ProtocolBuilder.builder(data))); - return true; - } catch (IOException e) { - // 发送失败 认为不可连接 + if (channelFuture != null) { + channelFuture.channel().writeAndFlush(data); + return true; + } + } catch (Exception e) { DataSenderFactoryWithBalance.unRegister(this); - return false; } + + return false; } public InetSocketAddress getServerIp() { return this.socketAddress; } - public void close(){ - if (socketChannel != null) { - try { - socketChannel.close(); - } catch (IOException e) { - logger.log(Level.ALL, "close connection Failed"); - } + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { + close(); + } + + public void close() { + if (group != null) { + group.shutdownGracefully(); } } diff --git a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java index f51614d04..58f1752e1 100644 --- a/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java +++ b/skywalking-api/src/main/java/com/ai/cloud/skywalking/sender/DataSenderFactoryWithBalance.java @@ -1,6 +1,7 @@ package com.ai.cloud.skywalking.sender; import com.ai.cloud.skywalking.conf.Config; +import com.ai.cloud.skywalking.protocol.Span; import com.ai.cloud.skywalking.util.StringUtil; import java.io.IOException; @@ -190,9 +191,9 @@ public class DataSenderFactoryWithBalance { DataSender result = null; int index = ThreadLocalRandom.current().nextInt(0, unusedServerAddresses.size()); - for (int i = 0; i < unusedServerAddresses.size();i++, index++) { + for (int i = 0; i < unusedServerAddresses.size(); i++, index++) { - if(index == unusedServerAddresses.size()){ + if (index == unusedServerAddresses.size()) { index = 0; } @@ -218,4 +219,11 @@ public class DataSenderFactoryWithBalance { .setStatus(DataSender.SenderStatus.FAILED); } } + + public static void main(String[] args) throws InterruptedException { + Span span = new Span("1.0a2.1453508782702.e8f7323.4115.762.256@~ @~0@~http://m.aisse.asiainfo.com/aisseMobilePage/toAisseMobilePage@~1453508782702@~85@~ITSC-MIS-LEV-web01/10.1.31.12@~1@~org.springframework.web.util.NestedServletException: Request processing failed; nested exception is java.lang.RuntimeException: org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLException: ORA-00257: archiver error. Connect internal only, until freed.#~#~org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLException: ORA-00257: archiver error. Connect internal only, until freed.#~#~\tat org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:241)#~\tat org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:372)#~\tat org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:417)#~\tat org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:255)#~\tat org.springframework.transaction.interceptor.TransactionInterceptor.invoke(TransactionInterceptor.java:94)#~\tat org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)#~\tat org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:204)#~\tat com.sun.proxy.$Proxy28.searchByNt(Unknown Source)#~\tat com.ai.aisse.core.rest.impl.ExpenseInitApiImpl.searchMembersinfo(ExpenseInitApiImpl.java:22)#~\tat com.alibaba.dubbo.common.bytecode.Wrapper3.invokeMethod(Wrapper3.java)#~\tat com.alibaba.dubbo.rpc.proxy.javassist.JavassistProxyFactory$1.doInvoke(JavassistProxyFactory.java:46)#~\tat com.alibaba.dubbo.rpc.proxy.AbstractProxyInvoker.invoke(AbstractProxyInvoker.java:72)#~\tat com.alibaba.dubbo.rpc.protocol.InvokerWrapper.invoke(InvokerWrapper.java:53)#~\tat com.ai.cloud.skywalking.plugin.dubbo.SWDubboEnhanceFilter.invoke(SWDubboEnhanceFilter.java:19)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.ExceptionFilter.invoke(ExceptionFilter.java:64)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.TimeoutFilter.invoke(TimeoutFilter.java:42)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.monitor.support.MonitorFilter.invoke(MonitorFilter.java:75)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.protocol.dubbo.filter.TraceFilter.invoke(TraceFilter.java:78)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.ContextFilter.invoke(ContextFilter.java:70)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.GenericFilter.invoke(GenericFilter.java:132)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.ClassLoaderFilter.invoke(ClassLoaderFilter.java:38)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.filter.EchoFilter.invoke(EchoFilter.java:38)#~\tat com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper$1.invoke(ProtocolFilterWrapper.java:91)#~\tat com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol$1.reply(DubboProtocol.java:113)#~\tat com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler.handleRequest(HeaderExchangeHandler.java:84)#~\tat com.alibaba.dubbo.remoting.exchange.support.header.HeaderExchangeHandler.received(HeaderExchangeHandler.java:170)#~\tat com.alibaba.dubbo.remoting.transport.DecodeHandler.received(DecodeHandler.java:52)#~\tat com.alibaba.dubbo.remoting.transport.dispatcher.ChannelEventRunnable.run(ChannelEventRunnable.java:82)#~\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)#~\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)#~\tat java.lang.Thread.run(Thread.java:745)#~Caused by: java.sql.SQLException: ORA-00257: archiver error. Connect internal only, until freed.#~#~\tat oracle.jdbc.driver.SQLStateMapping.newSQLException(SQLStateMapping.java:70)#~\tat oracle.jdbc.driver.DatabaseError.newSQLException(DatabaseError.java:112)#~\tat oracle.jdbc.driver.DatabaseError.throwSqlException(DatabaseError.java:173)#~\tat oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:455)#~\tat oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:406)#~\tat oracle.jdbc.driver.T4CTTIoer.processError(T4CTTIoer.java:399)#~\tat oracle.jdbc.driver.T4CTTIoauthenticate.receiveOsesskey(T4CTTIoauthenticate.java:306)#~\tat oracle.jdbc.driver.T4CConnection.logon(T4CConnection.java:387)#~\tat oracle.jdbc.driver.PhysicalConnection.(PhysicalConnection.java:490)#~\tat oracle.jdbc.driver.T4CConnection.(T4CConnection.java:202)#~\tat oracle.jdbc.driver.T4CDriverExtension.getConnection(T4CDriverExtension.java:33)#~\tat oracle.jdbc.driver.OracleDriver.connect(OracleDriver.java:474)#~\tat com.ai.cloud.skywalking.plugin.jdbc.TracingDriver.connect(TracingDriver.java:24)#~\tat org.apache.commons.dbcp2.DriverConnectionFactory.createConnection(DriverConnectionFactory.java:39)#~\tat org.apache.commons.dbcp2.PoolableConnectionFactory.makeObject(PoolableConnectionFactory.java:205)#~\tat org.apache.commons.pool2.impl.GenericObjectPool.create(GenericObjectPool.java:861)#~\tat org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:435)#~\tat org.apache.commons.pool2.impl.GenericObjectPool.borrowObject(GenericObjectPool.java:363)#~\tat org.apache.commons.dbcp2.PoolingDataSource.getConnection(PoolingDataSource.java:102)#~\tat org.apache.commons.dbcp2.BasicDataSource.getConnection(BasicDataSource.java:1413)#~\tat org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:203)#~\t... 38 more#~#~\tat org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:981)#~\tat org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:860)#~\tat javax.servlet.http.HttpServlet.service(HttpServlet.java:624)#~\tat org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:845)#~\tat javax.servlet.http.HttpServlet.service(HttpServlet.java:731)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:303)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat com.ai.sso.app.authentication.AuthenticationFilter.doFilter(AuthenticationFilter.java:70)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat com.ai.net.xss.filter.XSSFilter.doFilter(XSSFilter.java:38)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:121)#~\tat org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat com.ai.cloud.skywalking.plugin.web.SkyWalkingFilter.doFilter(SkyWalkingFilter.java:57)#~\tat org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)#~\tat org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)#~\tat org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:220)#~\tat org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:122)#~\tat org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:505)#~\tat org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:170)#~\tat com.googlecode.psiprobe.Tomcat70AgentValve.invoke(Tomcat70AgentValve.java:38)#~\tat org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:103)#~\tat org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:957)#~\tat org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:116)#~\tat org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:423)#~\tat org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1079)#~\tat org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:620)#~\tat org.apache.tomcat.util.net.AprEndpoint$SocketProcessor.doRun(AprEndpoint.java:2476)#~\tat org.apache.tomcat.util.net.AprEndpoint$SocketProcessor.run(AprEndpoint.java:2465)#~\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)#~\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)#~\tat org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)#~\tat java.lang.Thread.run(Thread.java:745)#~Caused by: java.lang.RuntimeException: org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLException: ORA-00257: archiver error. Connect internal only, until freed.#~#~org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLException: ORA-00257: archiver error. Connect internal only, until freed.#~#~\tat org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:241)#~\tat org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:372)#~\tat org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:417)#~\tat org.springframework.transaction.interceptor.TransactionAspectSupport.invokeWithinTransaction(TransactionAspectSupport.java:255)#~\tat org.springframewo \u0002.1.0a2.1453508787798.e8f7323.4115.1104.70@~0@~0@~com.ai.aisse.controller.myAisse.aisseMyselfPage.myAisse(com.ai.net.xss.wrapper.XssRequestWrapper,org.apache.catalina.connector.ResponseFacade,org.springframework.validation.support.BindingAwareModelMap)@~1453508787800@~4@~ITSC-MIS-LEV-web01/10.1.31.12@~0@~ @~M@~false@~ @~4115@~aisse-mobile-web@~5@~L"); + getSender().send(span.toString()); + + Thread.sleep(2000L); + } }