java - Curator 与 Zookeeper 服务发现连接被强制关闭

Question

我正在尝试使用 Curator 和 Zookeeper 进行服务发现。

当我尝试注册我的服务时，发生以下情况。

2014-09-23 16:36:09,755 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@197] - Accepted socket connection from /127.0.0.1:52026
2014-09-23 16:36:09,759 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@868] - Client attempting to establish new session at /127.0.0.1:52026
2014-09-23 16:36:09,790 [myid:] - INFO  [SyncThread:0:ZooKeeperServer@617] - Established session 0x148a17bad590007 with negotiated timeout 40000 for client /127.0.0.1:52026
2014-09-23 16:36:11,735 [myid:] - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@362] - Exception causing close of session 0x148a17bad590007 due to java.io.IOException: An existing con
nection was forcibly closed by the remote host
2014-09-23 16:36:11,736 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /127.0.0.1:52026 which had sessionid 0x148a17bad590007
2014-09-23 16:36:32,001 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x148a17bad590006, timeout of 40000ms exceeded
2014-09-23 16:36:32,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x148a17bad590006
2014-09-23 16:36:52,000 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x148a17bad590007, timeout of 40000ms exceeded
2014-09-23 16:36:52,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x148a17bad590007

相关代码如下:(导致上述错误的是getDiscovery方法)

private ServiceDiscovery<InstanceDetails> getDiscovery() {
        return ServiceDiscoveryBuilder.builder(InstanceDetails.class)
            .basePath(Config.basePath)
            .client(curatorFramework)
            .serializer(jacksonInstanceSerializer)
            .build();
    }

我在以下情况中使用它:

public void advertiseAvailability() {
    try {
        ServiceDiscovery<InstanceDetails> discovery = getDiscovery();
        discovery.start();
        discovery.registerService(getInstance()); // getInstance returns a ServiceInstance<InstanceDetails>
        discovery.close();
        System.out.println("Advertised");
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

以下是配置

public class Config{
    public static final String basePath = "/";
    public static final String serviceName = "myTestService1";
    public static final String address = "127.0.0.1";
    public static final int port = 2181;
}

以下是所需curatorFramework的初始化

    CuratorFramework curatorFramework;
    curatorFramework = CuratorFrameworkFactory.builder()
        .connectionTimeoutMs(1000)
        .retryPolicy(new RetryNTimes(10, 500))
        .connectString(Config.address+":"+Config.port)
        .build();
    curatorFramework.start();
    new EnsurePath(Config.basePath).ensure(curatorFramework.getZookeeperClient());

jacksonInstanceSerializer如下:

    this.jacksonInstanceSerializer = instanceSerializerFactory.getInstanceSerializer(
        new TypeReference<ServiceInstance<InstanceDetails>>() {}
    );

其中instanceSerializerFactory是:

public class InstanceSerializerFactory {
    private final ObjectReader objectReader;
    private final ObjectWriter objectWriter;

    public InstanceSerializerFactory(ObjectReader objectReader, ObjectWriter objectWriter) {
        this.objectReader = objectReader;
        this.objectWriter = objectWriter;
    }

    public <T> InstanceSerializer<T> getInstanceSerializer(
            TypeReference<ServiceInstance<T>> typeReference) {
        return new JacksonInstanceSerializer<T>(objectReader, objectWriter, typeReference);
    }

}

final class JacksonInstanceSerializer<T> implements InstanceSerializer<T> {
    private final TypeReference<ServiceInstance<T>> typeRef;
    private final ObjectWriter objectWriter;
    private final ObjectReader objectReader;

    JacksonInstanceSerializer(ObjectReader objectReader, ObjectWriter objectWriter,
                              TypeReference<ServiceInstance<T>> typeRef) {
        this.objectReader = objectReader;
        this.objectWriter = objectWriter;
        this.typeRef = typeRef;
    }

注意:实例详细说明了我尝试注册的内容(服务名称和地址等)

Answer 1

连接似乎正在从客户端关闭。 Zookeeper session 绑定(bind)到 Zookeeper 服务器和客户端之间的连接(在您的情况下使用 curator)。最有可能的是，客户端正在关闭，因此服务器正在识别这种情况并关闭您的 session 。