我正在处理莴苣集群Java客户端。它被设置在bolt
拓扑结构(Apache Strom)内。 spout
从kafka
读取数据并将其传递给bolt
。但是,当我启动我的拓扑时,我会收到以下错误消息并终止程序。我错过了什么吗?这是什么原因引起的?
堆栈跟踪
29502 [Thread-17-RecommendationLettuceBolt-executor[2 2]] ERROR o.a.s.util - Async loop died! io.lettuce.core.RedisException: Cannot retrieve initial cluster partitions from initial URIs [RedisURI [host='127.0.0.1', port=7001]] at io.lettuce.core.cluster.RedisClusterClient.loadPartitions(RedisClusterClient.java:865) at io.lettuce.core.cluster.RedisClusterClient.initializePartitions(RedisClusterClient.java:819) at io.lettuce.core.cluster.RedisClusterClient.connect(RedisClusterClient.java:345) at com.projectName.indexer.lettuce.LettuceClusterClientProvider.getConnection(LettuceClusterClientProvider.java:72) at com.projectName.indexer.lettuce.LettuceCacheRepopulationHandler.openLettuceConnection(LettuceCacheRepopulationHandler.java:42) at com.projectName.indexer.bolts.RecommendationLettuceBolt.prepare(RecommendationLettuceBolt.java:35) at org.apache.storm.daemon.executor$fn__8058$fn__8071.invoke(executor.clj:795) at org.apache.storm.util$async_loop$fn__624.invoke(util.clj:482) at clojure.lang.AFn.run(AFn.java:22) at java.base/java.lang.Thread.run(Thread.java:844) Caused by: io.lettuce.core.RedisConnectionException: Unable to establish a connection to Redis Cluster at [RedisURI [host='127.0.0.1', port=7001]] at io.lettuce.core.cluster.topology.AsyncConnections.get(AsyncConnections.java:89) at io.lettuce.core.cluster.topology.ClusterTopologyRefresh.loadViews(ClusterTopologyRefresh.java:73) at io.lettuce.core.cluster.RedisClusterClient.doLoadPartitions(RedisClusterClient.java:871) at io.lettuce.core.cluster.RedisClusterClient.loadPartitions(RedisClusterClient.java:844) ... 9 moreprivate void init() {
redisUri = RedisURI.Builder
.redis(lettuceConfig.getLettuceClusterHost())
.withPort(lettuceConfig.getLettuceClusterPort())
.withTimeout(Duration.ofMillis(lettuceConfig.getLettuceClusterTimeout()))
.build();
}
public StatefulRedisClusterConnection getConnection() {
if (connection == null || !connection.isOpen()) {
redisClusterClient = RedisClusterClient.create(redisUri);
final ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
.enablePeriodicRefresh(Duration.ofMinutes(BoltConstants.Lettuce.PERIODIC_REFRESH_TIME_IN_MIN))
.enableAdaptiveRefreshTrigger()
.build();
final ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
.autoReconnect(true)
.topologyRefreshOptions(topologyRefreshOptions)
.build();
redisClusterClient.setOptions(clusterClientOptions);
connection = redisClusterClient.connect(SnappyCompressor.wrap(new StringCodec()));
log.info("Connected to Redis client lettuce. lettuce connection is up and running.");
}
return connection;
}
环境
compile 'io.lettuce:lettuce-core:5.1.7.RELEASE'