트래픽을 복제하는 Netty 프록시

Question

Netty 프록시에서 두 개의 서버로 트래픽을 복제하는 방법을 모색 중입니다. 즉, 대신 일반적인 구현의 :

서버 1 -> 프록시 -> 서버 나 다음을 수행하고자하는 2

:

서버 1 -> 프록시 -> 서버 2 및 서버 3

Server3-> 프록시 삭제됨

각 메시지는 서버 2와 서버 3 모두로 전송됩니다.

프록시 서버와 서버 2 간의 통신이 서버 3 때문에 차단되지 않아야한다는 제약이 있습니다 (서버 3이 느린 경우 등) . https://github.com/dawnbreaks/TcpProxy

불행하게도, 내가 그물코 너무 익숙하지 않은,하지만 구현은 내 목적을 위해 고도로 최적화 보인다

다음 코드에서 시작하고있다. 이해하고 싶습니다 : 서버 3

에 통신 오버라이드 (override)하는 API

서버 3를위한 새로운 채널을 만드는 방법

1. 읽고 서버에서 메시지를 삭제하는 방법 3

Answer 1

님이 IRC#netty에서 채팅을 보았습니다.

여기 몇 가지가 있습니다. 프록시는 서버 1이 연결되는 서버 측을 가져야합니다. 그런 다음 서버 2와 서버 3은 프록시에서 연결을 제외하거나 프록시에서 데이터를 수신하기 위해 UDP (의존)를 사용해야합니다.

Netty에는 프록시 서버의 예가 있습니다. 이것은 당신의 경우에 효과적 일 것이고 그것의 제 3 부분을 위해 정말로 쉬울 것입니다. 간단히 기존 예제를 사용하고 서버 3에 연결될 새로운 연결을 열면됩니다. 이제 할 수있는 일은 프록시 (클라이언트 연결을 서버 2와 서버 3)에서 가져 오는 것입니다. 채널 그룹에 넣고 하나 작성하십시오 2 대의 서버까지! 내 예제 코드 편집은 ... 서버 1에서 서버 2로 프록시를 통해 통신을 허용하고 상호 대화를 허용하지만 서버 3 만 데이터를받을 수 있지만 서버 3이 프록시에 응답하면 프록시는 아무 것도하지 않습니다. . 버퍼를 해제하거나 서버 3에서 가져 오지 말아야하는 데이터를 처리하기 위해 핸들러를 추가 할 수도 있습니다. 또한 여기에서 시작해야하지만 netty 문서, api, 예제 및 ppt는 매우 유용합니다.

몇 가지 수정 된 코드를 첨부하여 알려 드리겠습니다. 여기에 예제 링크가 있습니다.

Netty Proxy Server Examples

은 그래서 예를 들어 당신은 HexDumpProxyFrontendHandler.class을 편집하는 단지 서버 3에 대한 새로운 클라이언트에 대한 두 번째 부트 스트랩을 추가합니다.

현재 코드

41  @Override 
42  public void channelActive(ChannelHandlerContext ctx) { 
43   final Channel inboundChannel = ctx.channel(); 
44 
45   // Start the connection attempt. 
46   Bootstrap b = new Bootstrap(); 
47   b.group(inboundChannel.eventLoop()) 
48   .channel(ctx.channel().getClass()) 
49   .handler(new HexDumpProxyBackendHandler(inboundChannel)) 
50   .option(ChannelOption.AUTO_READ, false); 
51   ChannelFuture f = b.connect(remoteHost, remotePort); 
52   outboundChannel = f.channel(); 
53   f.addListener(new ChannelFutureListener() { 
54    @Override 
55    public void operationComplete(ChannelFuture future) { 
56     if (future.isSuccess()) { 
57      // connection complete start to read first data 
58      inboundChannel.read(); 
59     } else { 
60      // Close the connection if the connection attempt has failed. 
61      inboundChannel.close(); 
62     } 
63    } 
64   }); 
65  } 

편집 코드

import io.netty.bootstrap.Bootstrap; 
import io.netty.channel.Channel; 
import io.netty.channel.ChannelFuture; 
import io.netty.channel.ChannelFutureListener; 
import io.netty.channel.ChannelHandlerContext; 
import io.netty.channel.ChannelOption; 

/* 
* Copyright 2012 The Netty Project 
* 
* The Netty Project licenses this file to you under the Apache License, 
* version 2.0 (the "License"); you may not use this file except in compliance 
* with the License. You may obtain a copy of the License at: 
* 
* http://www.apache.org/licenses/LICENSE-2.0 
* 
* Unless required by applicable law or agreed to in writing, software 
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 
* License for the specific language governing permissions and limitations 
* under the License. 
*/ 
package io.netty.example.proxy; 

import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelInboundHandlerAdapter; 
import io.netty.channel.group.ChannelGroup; 
import io.netty.channel.group.DefaultChannelGroup; 
import io.netty.util.concurrent.GlobalEventExecutor; 

public class HexDumpProxyFrontendHandler extends ChannelInboundHandlerAdapter { 

    private final String remoteHost; 
    private final int remotePort; 

    // As we use inboundChannel.eventLoop() when buildling the Bootstrap this does not need to be volatile as 
    // the server2OutboundChannel will use the same EventLoop (and therefore Thread) as the inboundChannel. 
    private Channel server2OutboundChannel; 
    private Channel server3OutboundChannel; 

    // TODO You should change this to your own executor 
    private ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE); 

    public HexDumpProxyFrontendHandler(String remoteHost, int remotePort) { 
     this.remoteHost = remoteHost; 
     this.remotePort = remotePort; 
    } 

    @Override 
    public void channelActive(ChannelHandlerContext ctx) { 
     final Channel inboundChannel = ctx.channel(); 

     // Start the connection attempt to SERVER 3 
     Bootstrap server3Bootstrap = new Bootstrap(); 
     server3Bootstrap.group(inboundChannel.eventLoop()) 
       .channel(ctx.channel().getClass()) 
       // You are only writing traffic to server 3 so you do not need to have a handler for the inbound traffic 
       .handler(new DiscardServerHandler()) // EDIT 
       .option(ChannelOption.AUTO_READ, false); 
     ChannelFuture server3Future = server3Bootstrap.connect(remoteHost, remotePort); 
     server3OutboundChannel = server3Future.channel(); 


     // Start the connection attempt to SERVER 2 
     Bootstrap server2Bootstrap = new Bootstrap(); 
     server2Bootstrap.group(inboundChannel.eventLoop()) 
       .channel(ctx.channel().getClass()) 
       .handler(new HexDumpProxyBackendHandler(inboundChannel)) 
       .option(ChannelOption.AUTO_READ, false); 
     ChannelFuture server2Future = server2Bootstrap.connect(remoteHost, remotePort); 
     server2OutboundChannel = server2Future.channel(); 
     server2Future.addListener(new ChannelFutureListener() { 
      @Override 
      public void operationComplete(ChannelFuture future) { 
       if (future.isSuccess()) { 
        // connection complete start to read first data 
        inboundChannel.read(); 
       } else { 
        // Close the connection if the connection attempt has failed. 
        inboundChannel.close(); 
       } 
      } 
     }); 

     // Here we are going to add channels to channel group to save bytebuf work 
     channels.add(server2OutboundChannel); 
     channels.add(server3OutboundChannel); 
    } 

    // You can keep this the same below or use the commented out section 
    @Override 
    public void channelRead(final ChannelHandlerContext ctx, Object msg) { 
     // You need to reference count the message +1 
     msg.retain(); 
     if (server2OutboundChannel.isActive()) { 
      server2OutboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() { 
       @Override 
       public void operationComplete(ChannelFuture future) { 
        if (future.isSuccess()) { 
         // was able to flush out data, start to read the next chunk 
         ctx.channel().read(); 
        } else { 
         future.channel().close(); 
        } 
       } 
      }); 
     } 
     if (server3OutboundChannel.isActive()) { 
      server3OutboundChannel.writeAndFlush(msg).addListener(new ChannelFutureListener() { 
       @Override 
       public void operationComplete(ChannelFuture future) { 
        if (future.isSuccess()) { 
         // was able to flush out data, start to read the next chunk 
         ctx.channel().read(); 
        } else { 
         future.channel().close(); 
        } 
       } 
      }); 
     } 


     // Optional to the above code instead channel writing automatically cares for reference counting for you 
//  channels.writeAndFlush(msg).addListeners(new ChannelFutureListener() { 
// 
//   @Override 
//   public void operationComplete(ChannelFuture future) throws Exception { 
//    if (future.isSuccess()) { 
//     // was able to flush out data, start to read the next chunk 
//     ctx.channel().read(); 
//    } else { 
//     future.channel().close(); 
//    } 
//   } 
//  }); 
    } 

    @Override 
    public void channelInactive(ChannelHandlerContext ctx) { 
     if (server2OutboundChannel != null) { 
      closeOnFlush(server2OutboundChannel); 
     } 
     if (server3OutboundChannel != null) { 
      closeOnFlush(server3OutboundChannel); 
     } 


     // Optionally can do this 
//  channels.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); 
    } 

    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
     cause.printStackTrace(); 
     closeOnFlush(ctx.channel()); 
    } 

    /** 
    * Closes the specified channel after all queued write requests are flushed. 
    */ 
    static void closeOnFlush(Channel ch) { 
     if (ch.isActive()) { 
      ch.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE); 
     } 
    } 
} 

폐기 처리기

서버 3에 의해 프록시에 기록 된 것을 버리기위한 처리기로서 서버 3에 추가 될 수 있습니다. 기본적으로 SimpleInboundHandlers는 참조 횟수를 감소시켜 처리 한 후 메시지를 삭제합니다.

Discard Handler Code

, 내가 서버 (3) 차단, 또는 느린 다운이되는 경우