如何在 Apache Camel 中检测网络断开连接? [英] How can I detect network disconnection in Apache Camel?

问题描述

当客户端与 Apache Camel 连接断开时，我需要登录.如何检测 netty-tcp 断开连接?

I need to log when a client is disconnected from Apache Camel connection. How can I detect a netty-tcp disconnection?

    from("{{uri.client.address}}")

推荐答案

记录单个客户端断开连接有点复杂，但我会尝试分享我必须做的事情.

Its a bit more complicated to log individual client disconnects but I will try to share what I had to do.

您需要使用您自己的类扩展 SimpleChannelInboundHandler 并添加所需的方法.有关示例类，请参见此处:

You need to extend SimpleChannelInboundHandler with your own class and add the methods needed. For an example class see here:

https://github.com/apache/camel/blob/master/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java

查看我已更改为包括客户端断开连接的主要方法是这个:

The main method there to look at which I have changed to include client disconnect is this one:

 @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
    //command to send when there is a disconnect from client side
    String command = "clientDisconnect";
    if (LOG.isTraceEnabled()) {
      LOG.trace("Channel closed: {}", ctx.channel());
    }
    LOG.info("Channel has disconnected: affects ", uniqueClientId);
    //create Exchange and let the consumer process it
    final Exchange exchange = consumer.getEndpoint().createExchange(ctx, command);
    //create inOnly as there is no client waiting for a response
    exchange.setPattern(ExchangePattern.InOnly);
    // set the exchange charset property for converting
    if (consumer.getConfiguration().getCharsetName() != null) {
      exchange.setProperty(Exchange.CHARSET_NAME, IOHelper.normalizeCharset(consumer.getConfiguration().getCharsetName()));
    }
    // we want to handle the UoW
    consumer.createUoW(exchange);
    //set the uniqueClientId
    beforeProcess(exchange, ctx, command);
    processAsynchronously(exchange, ctx, command);
    // to keep track of open sockets
    consumer.getNettyServerBootstrapFactory().removeChannel(ctx.channel());
    super.channelInactive(ctx);
}

如果频道已连接，请注意我记录的部分.

Note the part where I log if the channel has connected.

如果你想在你的主路由中为不同的客户端做不同的事情，你也可以记录不同客户端的IP地址.这应该对您有所帮助.

You can also log the ip address of different client if you want to do different things for different clients in your main route. This should help you.

扩展该类后，调用扩展类 TCPServerHandler.然后你需要创建一个 TCPServerHandlerFactory 来引用以前的类并设置你的编码器/解码器.这个类可以是这样的:

Once you have extended that class, call the extended class TCPServerHandler. Then you need to create a TCPServerHandlerFactory to refer to previous class and set your encoder/decoder. This class can look like this:

public class TCPServerInitializerFactory extends ServerInitializerFactory  {
  private NettyConsumer consumer;

  public TCPServerInitializerFactory(NettyConsumer consumer) {
    this.consumer = consumer;
  }
  @Override
  protected void initChannel(Channel channel) throws Exception {
    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addLast("string-encoder", new StringEncoder(CharsetUtil.UTF_8));
    pipeline.addLast("string-decoder", new StringDecoder(CharsetUtil.UTF_8));
    **pipeline.addLast("handler", new TCPServerHandler(consumer));**
  }

  @Override
  public ServerInitializerFactory createPipelineFactory(NettyConsumer nettyConsumer) {
    return new TCPServerInitializerFactory(nettyConsumer);
  }
}

标有 ** 的行是您创建 TCPServerHandler 类的位置.

The line marked with ** is where you are creating the TCPServerHandler class.

然后回到你的路由类中的骆驼，你需要像这样将工厂添加到你的注册表中:

Then back in camel in your route class you need to add the factory to your registry like this:

TCPServerInitializerFactory serverFactory = new TCPServerInitializerFactory(null);
    final CamelContext camelContext = getContext();
    final org.apache.camel.impl.SimpleRegistry registry = new org.apache.camel.impl.SimpleRegistry();
    final org.apache.camel.impl.CompositeRegistry compositeRegistry = new org.apache.camel.impl.CompositeRegistry();
    compositeRegistry.addRegistry(camelContext.getRegistry());
    compositeRegistry.addRegistry(registry);
    ((org.apache.camel.impl.DefaultCamelContext) camelContext).setRegistry(compositeRegistry);
registry.put("spf", serverFactory);

然后这样称呼它:

from("netty4:tcp://0.0.0.0:3000?serverInitializerFactory=#spf&sync=true")...

现在你可能想知道，所有这些都是为了注册一个断开连接的日志.这是因为要能够区分客户端何时断开连接并实际获取客户端的 ip 需要 Netty 代码.你可以在交换中保存一些价值，然后在你的路线中做一些事情.

Now you may wondering, all of this to register a log for a disconnect. This is because to be able to distinguish when a client disconnects and actually get the ip of the client you need to the Netty code. You can save some value in the exchange and then later in your route do something.

这篇关于如何在 Apache Camel 中检测网络断开连接?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！