我正在尝试使用 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;
}
注意:实例详细说明我要注册的内容(服务名称和地址等)