Wie kann ich mehrere HTTP-Anfragen asynchron mit Netty senden?

Question

Ich versuche, eine große Menge asynchron eine HTTP-Anfragen an einen Server zu senden. Mein Ziel ist es, jede Antwort mit ihrer ursprünglichen Anfrage zu vergleichen.

Dazu verfolge ich die Netty Snoop example.

In diesem Beispiel (und den anderen http-Beispielen) wird jedoch nicht behandelt, wie mehrere Anfragen asynchron gesendet werden oder wie diese anschließend mit den entsprechenden Anforderungen verknüpft werden.

Alle ähnlichen Fragen (wie this one, this one oder this one, die SimpleChannelUpstreamHandler Klasse implementieren, die von Netty 3 ist und existiert nicht mehr in 4,0 (documentation netty 4.0)

Wer hat eine Idee, wie diese zu lösen in netty 4,0

Edit:?

Mein Problem ist, obwohl ich viele Nachrichten an den Kanal zu schreiben, ich die Antworten nur sehr langsam erhalten (1 Antwort/sec, wHE reas eine Hoffnung, einige tausend/sec zu erhalten). Um dies zu verdeutlichen, bitte ich, was ich bisher bekommen habe. Ich bin sicher, dass der Server, an den ich die Anfragen sende, auch viel Verkehr bewältigen kann.

Was ich so weit gekommen:

import java.net.URI 
import java.nio.charset.StandardCharsets 
import java.io.File 

import io.netty.bootstrap.Bootstrap 
import io.netty.buffer.{Unpooled, ByteBuf} 
import io.netty.channel.{ChannelHandlerContext, SimpleChannelInboundHandler, ChannelInitializer} 
import io.netty.channel.socket.SocketChannel 
import io.netty.channel.socket.nio.NioSocketChannel 
import io.netty.handler.codec.http._ 
import io.netty.handler.timeout.IdleStateHandler 
import io.netty.util.{ReferenceCountUtil, CharsetUtil} 
import io.netty.channel.nio.NioEventLoopGroup 

import scala.io.Source 

object ClientTest { 

    val URL = System.getProperty("url", MY_URL)  
    val configuration = new Configuration 

    def main(args: Array[String]) { 
    println("Starting client") 
    start() 
    } 

    def start(): Unit = { 

    val group = new NioEventLoopGroup() 

    try { 

     val uri: URI = new URI(URL) 
     val host: String= {val h = uri.getHost(); if (h != null) h else "127.0.0.1"} 
     val port: Int = {val p = uri.getPort; if (p != -1) p else 80} 

     val b = new Bootstrap() 

     b.group(group) 
     .channel(classOf[NioSocketChannel]) 
     .handler(new HttpClientInitializer()) 

     val ch = b.connect(host, port).sync().channel() 

     val logFolder: File = new File(configuration.LOG_FOLDER) 
     val fileToProcess: Array[File] = logFolder.listFiles() 

     for (file <- fileToProcess){ 
     val name: String = file.getName() 
     val source = Source.fromFile(configuration.LOG_FOLDER + "/" + name) 

     val lineIterator: Iterator[String] = source.getLines() 

     while (lineIterator.hasNext) { 
      val line = lineIterator.next() 
      val jsonString = parseLine(line) 
      val request = createRequest(jsonString, uri, host) 
      ch.writeAndFlush(request) 
     } 
     println("closing") 
     ch.closeFuture().sync() 
     } 
    } finally { 
     group.shutdownGracefully() 
    } 
    } 

    private def parseLine(line: String) = { 
    //do some parsing to get the json string I want 
    } 

    def createRequest(jsonString: String, uri: URI, host: String): FullHttpRequest = { 
    val bytebuf: ByteBuf = Unpooled.copiedBuffer(jsonString, StandardCharsets.UTF_8) 

    val request: FullHttpRequest = new DefaultFullHttpRequest(
     HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath()) 
    request.headers().set(HttpHeaders.Names.HOST, host) 
    request.headers().set(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.KEEP_ALIVE) 
    request.headers().set(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP) 
    request.headers().add(HttpHeaders.Names.CONTENT_TYPE, "application/json") 

    request.headers().set(HttpHeaders.Names.CONTENT_LENGTH, bytebuf.readableBytes()) 
    request.content().clear().writeBytes(bytebuf) 

    request 
    } 
} 

class HttpClientInitializer() extends ChannelInitializer[SocketChannel] { 

    override def initChannel(ch: SocketChannel) = { 
    val pipeline = ch.pipeline() 

    pipeline.addLast(new HttpClientCodec()) 

    //aggregates all http messages into one if content is chunked 
    pipeline.addLast(new HttpObjectAggregator(1048576)) 

    pipeline.addLast(new IdleStateHandler(0, 0, 600)) 

    pipeline.addLast(new HttpClientHandler()) 
    } 
} 

class HttpClientHandler extends SimpleChannelInboundHandler[HttpObject] { 

    override def channelRead0(ctx: ChannelHandlerContext, msg: HttpObject) { 
    try { 
     msg match { 
     case res: FullHttpResponse => 
      println("response is: " + res.content().toString(CharsetUtil.US_ASCII)) 
      ReferenceCountUtil.retain(msg) 
     } 
    } finally { 
     ReferenceCountUtil.release(msg) 
    } 
    } 

    override def exceptionCaught(ctx: ChannelHandlerContext, e: Throwable) = { 
    println("HttpHandler caught exception", e) 
    ctx.close() 
    } 
} 

Answer 1

ChannelFuture cf = channel.writeAndFlush (erzeugeAnf());

noch wie sie später mit den entsprechenden Anfragen verknüpft werden.

Can netty assign multiple IO threads to the same Channel?

der Arbeiter-Thread einmal für einen zugewiesenen Kanal für die gesamte Lebensdauer des Kanals sich nicht ändert. Wir profitieren also nicht von den Threads. Das liegt daran, dass Sie die Verbindung aufrechterhalten und der Kanal am Leben erhalten wird.

Um dieses Problem zu beheben, könnten Sie einen Pool von Kanälen in Betracht ziehen (sagen wir 30). Verwenden Sie dann den Kanalpool, um Ihre Anforderungen zu platzieren.

 int concurrent = 30; 

    // Start the client. 
    ChannelFuture[] channels = new ChannelFuture[concurrent]; 
    for (int i = 0; i < channels.length; i++) { 
    channels[i] = b.connect(host, port).sync(); 
    } 

    for (int i = 0; i < 1000; i++) { 
     ChannelFuture requestHandle = process(channels[(i+1)%concurrent]); 
     // do something with the request handle  
    } 

    for (int i = 0; i < channels.length; i++) { 
    channels[i].channel().closeFuture().sync(); 
    } 

HTH