基于Storm的日志收集系统

/** * Copyright : Copyright (c) 2006. Wintim Corp. All rights reserved * File Summary : class of connector through netty * Author : wuyadong * Create time : 2013-2-21 * Project Name : Sailfish1.4.3 */ package com.wintim.connect; import java.net.InetSocketAddress; import java.util.List; import java.util.Random; import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import org.jboss.netty.bootstrap.ClientBootstrap; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelFuture; import org.jboss.netty.channel.ChannelFutureListener; import org.jboss.netty.channel.ChannelHandlerContext; import org.jboss.netty.channel.ChannelPipeline; import org.jboss.netty.channel.ChannelPipelineFactory; import org.jboss.netty.channel.ChannelStateEvent; import org.jboss.netty.channel.Channels; import org.jboss.netty.channel.ExceptionEvent; import org.jboss.netty.channel.SimpleChannelUpstreamHandler; import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory; import org.jboss.netty.handler.codec.serialization.ObjectEncoder; import com.wintim.buffer.IFlushEvent; import com.wintim.observer.IEvent; /*** * communicate with server through netty * */ public class NettyConnector implements IConnect { private static NettyConnector connector = null; private static List<Object[]> addressList = null; private static int formerServerIndex = -1; private ClientBootstrap clientBootstrap = null; private Channel channel = null; private AtomicBoolean isConnected = null; private AtomicBoolean isConnecting = null; private NettyConnector(String ip, int port) { isConnected = new AtomicBoolean(false); isConnecting = new AtomicBoolean(false); clientBootstrap = new ClientBootstrap(new NioClientSocketChannelFactory( Executors.newCachedThreadPool(), Executors.newCachedThreadPool())); clientBootstrap.setPipelineFactory(new ChannelPipelineFactory() { @Override public ChannelPipeline getPipeline() throws Exception { // TODO Auto-generated method stub ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("log-decoder", new ObjectEncoder()); pipeline.addLast("command-handler", new SimpleChannelUpstreamHandler(){ @Override public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { // TODO Auto-generated method stub isConnected.set(true); channel = e.getChannel(); super.channelConnected(ctx, e); } @Override public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { // TODO Auto-generated method stub isConnected.set(false); System.out.println("netty connector: channel closed!"); super.channelClosed(ctx, e); } @Override public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception { // TODO Auto-generated method stub isConnected.set(false); System.out.println("nettyConnector: channel disconnected!"); super.channelDisconnected(ctx, e); } @Override public void exceptionCaught(ChannelHandlerContext arg0, ExceptionEvent exception) throws Exception { // TODO Auto-generated method stub System.err.println(exception); } }); return pipeline; } }); clientBootstrap.setOption("tcpNoDelay", false); clientBootstrap.setOption("keepAlive", false); clientBootstrap.setOption("reuseAddress", true); clientBootstrap.setOption("soLinger", 0); clientBootstrap.setOption("remoteAddress", new InetSocketAddress(ip, port)); clientBootstrap.setOption("connectTimeoutMillis", 500); System.out.println("netty connector init finished!"); } private static NettyConnector getInstance(String ip, int port) { if (connector == null) { connector = new NettyConnector(ip, port); } return connector; } public static NettyConnector getInstance(List<Object[]> list) { if (list == null) { throw new NullPointerException("address list can't be null"); } if (list.size() <= 0) { throw new IllegalArgumentException("address list can't be empty!"); } if (addressList == null) { addressList = list; } Object[] object = getAddress(); return getInstance((String)object[0],(Integer)object[1]); } //random server, except former server private static Object[] getAddress() { Random random = new Random(System.currentTimeMillis()); int i = 0; if (addressList.size() > 1) { i = random.nextInt(addressList.size() - 1); //if i == former server ,then add 1 if (i == formerServerIndex) { i = (formerServerIndex + 1) % addressList.size(); } } else if (addressList.size() == 1) { i = 0; } return addressList.get(i); } @Override public synchronized boolean send(Object obj) { // TODO Auto-generated method stub if (channel != null && isConnected.get() && channel.isWritable()) { channel.write(obj); return true; } else { return false; } } @Override public void connect() { // TODO Auto-generated method stub if (clientBootstrap != null ) { //if it's locked, then do nothing, just get out! synchronized (NettyConnector.this) { if (isConnecting.get() || isConnected.get()) return ; isConnecting.set(true); } //change server Object[] object = getAddress(); System.out.println("connect..."); System.out.println("address:" + object[0] + ":" + object[1]); clientBootstrap.setOption("remoteAddress", new InetSocketAddress((String)object[0], (Integer)object[1])); ChannelFuture future = clientBootstrap.connect(); future.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture f) throws Exception { // TODO Auto-generated method stub if (f.isSuccess()) { isConnected.set(true); System.out.println("connected!"); } else { isConnected.set(false); System.err.println("unconnected!"); } isConnecting.set(false); } }); } } @Override public void close() { // TODO Auto-generated method stub if (clientBootstrap != null) { if (channel != null) { System.out.println("close"); channel.close().awaitUninterruptibly(); //channel.getCloseFuture().awaitUninterruptibly(); } //clientBootstrap.releaseExternalResources(); } } /**** * notice ,whether it is connected, depended on * the former result of connection. * * */ @Override public boolean isConnected() { // TODO Auto-generated method stub if (isConnected.get() && channel != null) { return true; } else { return false; } } @Override public void update(IEvent event) { // TODO Auto-generated method stub if (event.getCommand().equals(IFlushEvent.BUFFER_FLUSH_EVENT)) { //if can connected ,send them to server if (isConnected()) { send(event.getContent()); } else { //if cann't store them in sailfish.log System.err.println("Can't Send log"); System.err.println(event.getContent().get("logs")); } } } }